  <html>
    <head>
      <meta charset="utf-8">
      <title>Spark core InsightCreate Repository</title>
      <style>
        #wrapper {width: 960px; margin: 0 auto;}
        /* Asciidoctor default stylesheet | MIT License | http://asciidoctor.org */
/* Uncomment @import statement below to use as custom stylesheet */
/*@import "https://fonts.googleapis.com/css?family=Open+Sans:300,300italic,400,400italic,600,600italic%7CNoto+Serif:400,400italic,700,700italic%7CDroid+Sans+Mono:400,700";*/
article,aside,details,figcaption,figure,footer,header,hgroup,main,nav,section,summary{display:block}
audio,canvas,video{display:inline-block}
audio:not([controls]){display:none;height:0}
script{display:none!important}
html{font-family:sans-serif;-ms-text-size-adjust:100%;-webkit-text-size-adjust:100%}
a{background:transparent}
a:focus{outline:thin dotted}
a:active,a:hover{outline:0}
h1{font-size:2em;margin:.67em 0}
abbr[title]{border-bottom:1px dotted}
b,strong{font-weight:bold}
dfn{font-style:italic}
hr{-moz-box-sizing:content-box;box-sizing:content-box;height:0}
mark{background:#ff0;color:#000}
code,kbd,pre,samp{font-family:monospace;font-size:1em}
pre{white-space:pre-wrap}
q{quotes:"\201C" "\201D" "\2018" "\2019"}
small{font-size:80%}
sub,sup{font-size:75%;line-height:0;position:relative;vertical-align:baseline}
sup{top:-.5em}
sub{bottom:-.25em}
img{border:0}
svg:not(:root){overflow:hidden}
figure{margin:0}
fieldset{border:1px solid silver;margin:0 2px;padding:.35em .625em .75em}
legend{border:0;padding:0}
button,input,select,textarea{font-family:inherit;font-size:100%;margin:0}
button,input{line-height:normal}
button,select{text-transform:none}
button,html input[type="button"],input[type="reset"],input[type="submit"]{-webkit-appearance:button;cursor:pointer}
button[disabled],html input[disabled]{cursor:default}
input[type="checkbox"],input[type="radio"]{box-sizing:border-box;padding:0}
button::-moz-focus-inner,input::-moz-focus-inner{border:0;padding:0}
textarea{overflow:auto;vertical-align:top}
table{border-collapse:collapse;border-spacing:0}
*,*::before,*::after{-moz-box-sizing:border-box;-webkit-box-sizing:border-box;box-sizing:border-box}
html,body{font-size:100%}
body{background:#fff;color:rgba(0,0,0,.8);padding:0;margin:0;font-family:"Noto Serif","DejaVu Serif",serif;font-weight:400;font-style:normal;line-height:1;position:relative;cursor:auto;tab-size:4;-moz-osx-font-smoothing:grayscale;-webkit-font-smoothing:antialiased}
a:hover{cursor:pointer}
img,object,embed{max-width:100%;height:auto}
object,embed{height:100%}
img{-ms-interpolation-mode:bicubic}
.left{float:left!important}
.right{float:right!important}
.text-left{text-align:left!important}
.text-right{text-align:right!important}
.text-center{text-align:center!important}
.text-justify{text-align:justify!important}
.hide{display:none}
img,object,svg{display:inline-block;vertical-align:middle}
textarea{height:auto;min-height:50px}
select{width:100%}
.center{margin-left:auto;margin-right:auto}
.stretch{width:100%}
.subheader,.admonitionblock td.content>.title,.audioblock>.title,.exampleblock>.title,.imageblock>.title,.listingblock>.title,.literalblock>.title,.stemblock>.title,.openblock>.title,.paragraph>.title,.quoteblock>.title,table.tableblock>.title,.verseblock>.title,.videoblock>.title,.dlist>.title,.olist>.title,.ulist>.title,.qlist>.title,.hdlist>.title{line-height:1.45;color:#7a2518;font-weight:400;margin-top:0;margin-bottom:.25em}
div,dl,dt,dd,ul,ol,li,h1,h2,h3,#toctitle,.sidebarblock>.content>.title,h4,h5,h6,pre,form,p,blockquote,th,td{margin:0;padding:0;direction:ltr}
a{color:#2156a5;text-decoration:underline;line-height:inherit}
a:hover,a:focus{color:#1d4b8f}
a img{border:none}
p{font-family:inherit;font-weight:400;font-size:1em;line-height:1.6;margin-bottom:1.25em;text-rendering:optimizeLegibility}
p aside{font-size:.875em;line-height:1.35;font-style:italic}
h1,h2,h3,#toctitle,.sidebarblock>.content>.title,h4,h5,h6{font-family:"Open Sans","DejaVu Sans",sans-serif;font-weight:300;font-style:normal;color:#ba3925;text-rendering:optimizeLegibility;margin-top:1em;margin-bottom:.5em;line-height:1.0125em}
h1 small,h2 small,h3 small,#toctitle small,.sidebarblock>.content>.title small,h4 small,h5 small,h6 small{font-size:60%;color:#e99b8f;line-height:0}
h1{font-size:2.125em}
h2{font-size:1.6875em}
h3,#toctitle,.sidebarblock>.content>.title{font-size:1.375em}
h4,h5{font-size:1.125em}
h6{font-size:1em}
hr{border:solid #dddddf;border-width:1px 0 0;clear:both;margin:1.25em 0 1.1875em;height:0}
em,i{font-style:italic;line-height:inherit}
strong,b{font-weight:bold;line-height:inherit}
small{font-size:60%;line-height:inherit}
code{font-family:"Droid Sans Mono","DejaVu Sans Mono",monospace;font-weight:400;color:rgba(0,0,0,.9)}
ul,ol,dl{font-size:1em;line-height:1.6;margin-bottom:1.25em;list-style-position:outside;font-family:inherit}
ul,ol{margin-left:1.5em}
ul li ul,ul li ol{margin-left:1.25em;margin-bottom:0;font-size:1em}
ul.square li ul,ul.circle li ul,ul.disc li ul{list-style:inherit}
ul.square{list-style-type:square}
ul.circle{list-style-type:circle}
ul.disc{list-style-type:disc}
ol li ul,ol li ol{margin-left:1.25em;margin-bottom:0}
dl dt{margin-bottom:.3125em;font-weight:bold}
dl dd{margin-bottom:1.25em}
abbr,acronym{text-transform:uppercase;font-size:90%;color:rgba(0,0,0,.8);border-bottom:1px dotted #ddd;cursor:help}
abbr{text-transform:none}
blockquote{margin:0 0 1.25em;padding:.5625em 1.25em 0 1.1875em;border-left:1px solid #ddd}
blockquote cite{display:block;font-size:.9375em;color:rgba(0,0,0,.6)}
blockquote cite::before{content:"\2014 \0020"}
blockquote cite a,blockquote cite a:visited{color:rgba(0,0,0,.6)}
blockquote,blockquote p{line-height:1.6;color:rgba(0,0,0,.85)}
@media screen and (min-width:768px){h1,h2,h3,#toctitle,.sidebarblock>.content>.title,h4,h5,h6{line-height:1.2}
h1{font-size:2.75em}
h2{font-size:2.3125em}
h3,#toctitle,.sidebarblock>.content>.title{font-size:1.6875em}
h4{font-size:1.4375em}}
table{background:#fff;margin-bottom:1.25em;border:solid 1px #dedede}
table thead,table tfoot{background:#f7f8f7}
table thead tr th,table thead tr td,table tfoot tr th,table tfoot tr td{padding:.5em .625em .625em;font-size:inherit;color:rgba(0,0,0,.8);text-align:left}
table tr th,table tr td{padding:.5625em .625em;font-size:inherit;color:rgba(0,0,0,.8)}
table tr.even,table tr.alt,table tr:nth-of-type(even){background:#f8f8f7}
table thead tr th,table tfoot tr th,table tbody tr td,table tr td,table tfoot tr td{display:table-cell;line-height:1.6}
h1,h2,h3,#toctitle,.sidebarblock>.content>.title,h4,h5,h6{line-height:1.2;word-spacing:-.05em}
h1 strong,h2 strong,h3 strong,#toctitle strong,.sidebarblock>.content>.title strong,h4 strong,h5 strong,h6 strong{font-weight:400}
.clearfix::before,.clearfix::after,.float-group::before,.float-group::after{content:" ";display:table}
.clearfix::after,.float-group::after{clear:both}
*:not(pre)>code{font-size:.9375em;font-style:normal!important;letter-spacing:0;padding:.1em .5ex;word-spacing:-.15em;background-color:#f7f7f8;-webkit-border-radius:4px;border-radius:4px;line-height:1.45;text-rendering:optimizeSpeed;word-wrap:break-word}
*:not(pre)>code.nobreak{word-wrap:normal}
*:not(pre)>code.nowrap{white-space:nowrap}
pre,pre>code{line-height:1.45;color:rgba(0,0,0,.9);font-family:"Droid Sans Mono","DejaVu Sans Mono",monospace;font-weight:400;text-rendering:optimizeSpeed}
em em{font-style:normal}
strong strong{font-weight:400}
.keyseq{color:rgba(51,51,51,.8)}
kbd{font-family:"Droid Sans Mono","DejaVu Sans Mono",monospace;display:inline-block;color:rgba(0,0,0,.8);font-size:.65em;line-height:1.45;background-color:#f7f7f7;border:1px solid #ccc;-webkit-border-radius:3px;border-radius:3px;-webkit-box-shadow:0 1px 0 rgba(0,0,0,.2),0 0 0 .1em white inset;box-shadow:0 1px 0 rgba(0,0,0,.2),0 0 0 .1em #fff inset;margin:0 .15em;padding:.2em .5em;vertical-align:middle;position:relative;top:-.1em;white-space:nowrap}
.keyseq kbd:first-child{margin-left:0}
.keyseq kbd:last-child{margin-right:0}
.menuseq,.menuref{color:#000}
.menuseq b:not(.caret),.menuref{font-weight:inherit}
.menuseq{word-spacing:-.02em}
.menuseq b.caret{font-size:1.25em;line-height:.8}
.menuseq i.caret{font-weight:bold;text-align:center;width:.45em}
b.button::before,b.button::after{position:relative;top:-1px;font-weight:400}
b.button::before{content:"[";padding:0 3px 0 2px}
b.button::after{content:"]";padding:0 2px 0 3px}
p a>code:hover{color:rgba(0,0,0,.9)}
#header,#content,#footnotes,#footer{width:100%;margin-left:auto;margin-right:auto;margin-top:0;margin-bottom:0;max-width:62.5em;*zoom:1;position:relative;padding-left:.9375em;padding-right:.9375em}
#header::before,#header::after,#content::before,#content::after,#footnotes::before,#footnotes::after,#footer::before,#footer::after{content:" ";display:table}
#header::after,#content::after,#footnotes::after,#footer::after{clear:both}
#content{margin-top:1.25em}
#content::before{content:none}
#header>h1:first-child{color:rgba(0,0,0,.85);margin-top:2.25rem;margin-bottom:0}
#header>h1:first-child+#toc{margin-top:8px;border-top:1px solid #dddddf}
#header>h1:only-child,body.toc2 #header>h1:nth-last-child(2){border-bottom:1px solid #dddddf;padding-bottom:8px}
#header .details{border-bottom:1px solid #dddddf;line-height:1.45;padding-top:.25em;padding-bottom:.25em;padding-left:.25em;color:rgba(0,0,0,.6);display:-ms-flexbox;display:-webkit-flex;display:flex;-ms-flex-flow:row wrap;-webkit-flex-flow:row wrap;flex-flow:row wrap}
#header .details span:first-child{margin-left:-.125em}
#header .details span.email a{color:rgba(0,0,0,.85)}
#header .details br{display:none}
#header .details br+span::before{content:"\00a0\2013\00a0"}
#header .details br+span.author::before{content:"\00a0\22c5\00a0";color:rgba(0,0,0,.85)}
#header .details br+span#revremark::before{content:"\00a0|\00a0"}
#header #revnumber{text-transform:capitalize}
#header #revnumber::after{content:"\00a0"}
#content>h1:first-child:not([class]){color:rgba(0,0,0,.85);border-bottom:1px solid #dddddf;padding-bottom:8px;margin-top:0;padding-top:1rem;margin-bottom:1.25rem}
#toc{border-bottom:1px solid #e7e7e9;padding-bottom:.5em}
#toc>ul{margin-left:.125em}
#toc ul.sectlevel0>li>a{font-style:italic}
#toc ul.sectlevel0 ul.sectlevel1{margin:.5em 0}
#toc ul{font-family:"Open Sans","DejaVu Sans",sans-serif;list-style-type:none}
#toc li{line-height:1.3334;margin-top:.3334em}
#toc a{text-decoration:none}
#toc a:active{text-decoration:underline}
#toctitle{color:#7a2518;font-size:1.2em}
@media screen and (min-width:768px){#toctitle{font-size:1.375em}
body.toc2{padding-left:15em;padding-right:0}
#toc.toc2{margin-top:0!important;background-color:#f8f8f7;position:fixed;width:15em;left:0;top:0;border-right:1px solid #e7e7e9;border-top-width:0!important;border-bottom-width:0!important;z-index:1000;padding:1.25em 1em;height:100%;overflow:auto}
#toc.toc2 #toctitle{margin-top:0;margin-bottom:.8rem;font-size:1.2em}
#toc.toc2>ul{font-size:.9em;margin-bottom:0}
#toc.toc2 ul ul{margin-left:0;padding-left:1em}
#toc.toc2 ul.sectlevel0 ul.sectlevel1{padding-left:0;margin-top:.5em;margin-bottom:.5em}
body.toc2.toc-right{padding-left:0;padding-right:15em}
body.toc2.toc-right #toc.toc2{border-right-width:0;border-left:1px solid #e7e7e9;left:auto;right:0}}
@media screen and (min-width:1280px){body.toc2{padding-left:20em;padding-right:0}
#toc.toc2{width:20em}
#toc.toc2 #toctitle{font-size:1.375em}
#toc.toc2>ul{font-size:.95em}
#toc.toc2 ul ul{padding-left:1.25em}
body.toc2.toc-right{padding-left:0;padding-right:20em}}
#content #toc{border-style:solid;border-width:1px;border-color:#e0e0dc;margin-bottom:1.25em;padding:1.25em;background:#f8f8f7;-webkit-border-radius:4px;border-radius:4px}
#content #toc>:first-child{margin-top:0}
#content #toc>:last-child{margin-bottom:0}
#footer{max-width:100%;background-color:rgba(0,0,0,.8);padding:1.25em}
#footer-text{color:rgba(255,255,255,.8);line-height:1.44}
#content{margin-bottom:.625em}
.sect1{padding-bottom:.625em}
@media screen and (min-width:768px){#content{margin-bottom:1.25em}
.sect1{padding-bottom:1.25em}}
.sect1:last-child{padding-bottom:0}
.sect1+.sect1{border-top:1px solid #e7e7e9}
#content h1>a.anchor,h2>a.anchor,h3>a.anchor,#toctitle>a.anchor,.sidebarblock>.content>.title>a.anchor,h4>a.anchor,h5>a.anchor,h6>a.anchor{position:absolute;z-index:1001;width:1.5ex;margin-left:-1.5ex;display:block;text-decoration:none!important;visibility:hidden;text-align:center;font-weight:400}
#content h1>a.anchor::before,h2>a.anchor::before,h3>a.anchor::before,#toctitle>a.anchor::before,.sidebarblock>.content>.title>a.anchor::before,h4>a.anchor::before,h5>a.anchor::before,h6>a.anchor::before{content:"\00A7";font-size:.85em;display:block;padding-top:.1em}
#content h1:hover>a.anchor,#content h1>a.anchor:hover,h2:hover>a.anchor,h2>a.anchor:hover,h3:hover>a.anchor,#toctitle:hover>a.anchor,.sidebarblock>.content>.title:hover>a.anchor,h3>a.anchor:hover,#toctitle>a.anchor:hover,.sidebarblock>.content>.title>a.anchor:hover,h4:hover>a.anchor,h4>a.anchor:hover,h5:hover>a.anchor,h5>a.anchor:hover,h6:hover>a.anchor,h6>a.anchor:hover{visibility:visible}
#content h1>a.link,h2>a.link,h3>a.link,#toctitle>a.link,.sidebarblock>.content>.title>a.link,h4>a.link,h5>a.link,h6>a.link{color:#ba3925;text-decoration:none}
#content h1>a.link:hover,h2>a.link:hover,h3>a.link:hover,#toctitle>a.link:hover,.sidebarblock>.content>.title>a.link:hover,h4>a.link:hover,h5>a.link:hover,h6>a.link:hover{color:#a53221}
.audioblock,.imageblock,.literalblock,.listingblock,.stemblock,.videoblock{margin-bottom:1.25em}
.admonitionblock td.content>.title,.audioblock>.title,.exampleblock>.title,.imageblock>.title,.listingblock>.title,.literalblock>.title,.stemblock>.title,.openblock>.title,.paragraph>.title,.quoteblock>.title,table.tableblock>.title,.verseblock>.title,.videoblock>.title,.dlist>.title,.olist>.title,.ulist>.title,.qlist>.title,.hdlist>.title{text-rendering:optimizeLegibility;text-align:left;font-family:"Noto Serif","DejaVu Serif",serif;font-size:1rem;font-style:italic}
table.tableblock.fit-content>caption.title{white-space:nowrap;width:0}
.paragraph.lead>p,#preamble>.sectionbody>[class="paragraph"]:first-of-type p{font-size:1.21875em;line-height:1.6;color:rgba(0,0,0,.85)}
table.tableblock #preamble>.sectionbody>[class="paragraph"]:first-of-type p{font-size:inherit}
.admonitionblock>table{border-collapse:separate;border:0;background:none;width:100%}
.admonitionblock>table td.icon{text-align:center;width:80px}
.admonitionblock>table td.icon img{max-width:none}
.admonitionblock>table td.icon .title{font-weight:bold;font-family:"Open Sans","DejaVu Sans",sans-serif;text-transform:uppercase}
.admonitionblock>table td.content{padding-left:1.125em;padding-right:1.25em;border-left:1px solid #dddddf;color:rgba(0,0,0,.6)}
.admonitionblock>table td.content>:last-child>:last-child{margin-bottom:0}
.exampleblock>.content{border-style:solid;border-width:1px;border-color:#e6e6e6;margin-bottom:1.25em;padding:1.25em;background:#fff;-webkit-border-radius:4px;border-radius:4px}
.exampleblock>.content>:first-child{margin-top:0}
.exampleblock>.content>:last-child{margin-bottom:0}
.sidebarblock{border-style:solid;border-width:1px;border-color:#e0e0dc;margin-bottom:1.25em;padding:1.25em;background:#f8f8f7;-webkit-border-radius:4px;border-radius:4px}
.sidebarblock>:first-child{margin-top:0}
.sidebarblock>:last-child{margin-bottom:0}
.sidebarblock>.content>.title{color:#7a2518;margin-top:0;text-align:center}
.exampleblock>.content>:last-child>:last-child,.exampleblock>.content .olist>ol>li:last-child>:last-child,.exampleblock>.content .ulist>ul>li:last-child>:last-child,.exampleblock>.content .qlist>ol>li:last-child>:last-child,.sidebarblock>.content>:last-child>:last-child,.sidebarblock>.content .olist>ol>li:last-child>:last-child,.sidebarblock>.content .ulist>ul>li:last-child>:last-child,.sidebarblock>.content .qlist>ol>li:last-child>:last-child{margin-bottom:0}
.literalblock pre,.listingblock pre:not(.highlight),.listingblock pre[class="highlight"],.listingblock pre[class^="highlight "],.listingblock pre.CodeRay,.listingblock pre.prettyprint{background:#f7f7f8}
.sidebarblock .literalblock pre,.sidebarblock .listingblock pre:not(.highlight),.sidebarblock .listingblock pre[class="highlight"],.sidebarblock .listingblock pre[class^="highlight "],.sidebarblock .listingblock pre.CodeRay,.sidebarblock .listingblock pre.prettyprint{background:#f2f1f1}
.literalblock pre,.literalblock pre[class],.listingblock pre,.listingblock pre[class]{-webkit-border-radius:4px;border-radius:4px;word-wrap:break-word;overflow-x:auto;padding:1em;font-size:.8125em}
@media screen and (min-width:768px){.literalblock pre,.literalblock pre[class],.listingblock pre,.listingblock pre[class]{font-size:.90625em}}
@media screen and (min-width:1280px){.literalblock pre,.literalblock pre[class],.listingblock pre,.listingblock pre[class]{font-size:1em}}
.literalblock pre.nowrap,.literalblock pre.nowrap pre,.listingblock pre.nowrap,.listingblock pre.nowrap pre{white-space:pre;word-wrap:normal}
.literalblock.output pre{color:#f7f7f8;background-color:rgba(0,0,0,.9)}
.listingblock pre.highlightjs{padding:0}
.listingblock pre.highlightjs>code{padding:1em;-webkit-border-radius:4px;border-radius:4px}
.listingblock pre.prettyprint{border-width:0}
.listingblock>.content{position:relative}
.listingblock code[data-lang]::before{display:none;content:attr(data-lang);position:absolute;font-size:.75em;top:.425rem;right:.5rem;line-height:1;text-transform:uppercase;color:#999}
.listingblock:hover code[data-lang]::before{display:block}
.listingblock.terminal pre .command::before{content:attr(data-prompt);padding-right:.5em;color:#999}
.listingblock.terminal pre .command:not([data-prompt])::before{content:"$"}
table.pyhltable{border-collapse:separate;border:0;margin-bottom:0;background:none}
table.pyhltable td{vertical-align:top;padding-top:0;padding-bottom:0;line-height:1.45}
table.pyhltable td.code{padding-left:.75em;padding-right:0}
pre.pygments .lineno,table.pyhltable td:not(.code){color:#999;padding-left:0;padding-right:.5em;border-right:1px solid #dddddf}
pre.pygments .lineno{display:inline-block;margin-right:.25em}
table.pyhltable .linenodiv{background:none!important;padding-right:0!important}
.quoteblock{margin:0 1em 1.25em 1.5em;display:table}
.quoteblock>.title{margin-left:-1.5em;margin-bottom:.75em}
.quoteblock blockquote,.quoteblock p{color:rgba(0,0,0,.85);font-size:1.15rem;line-height:1.75;word-spacing:.1em;letter-spacing:0;font-style:italic;text-align:justify}
.quoteblock blockquote{margin:0;padding:0;border:0}
.quoteblock blockquote::before{content:"\201c";float:left;font-size:2.75em;font-weight:bold;line-height:.6em;margin-left:-.6em;color:#7a2518;text-shadow:0 1px 2px rgba(0,0,0,.1)}
.quoteblock blockquote>.paragraph:last-child p{margin-bottom:0}
.quoteblock .attribution{margin-top:.75em;margin-right:.5ex;text-align:right}
.verseblock{margin:0 1em 1.25em}
.verseblock pre{font-family:"Open Sans","DejaVu Sans",sans;font-size:1.15rem;color:rgba(0,0,0,.85);font-weight:300;text-rendering:optimizeLegibility}
.verseblock pre strong{font-weight:400}
.verseblock .attribution{margin-top:1.25rem;margin-left:.5ex}
.quoteblock .attribution,.verseblock .attribution{font-size:.9375em;line-height:1.45;font-style:italic}
.quoteblock .attribution br,.verseblock .attribution br{display:none}
.quoteblock .attribution cite,.verseblock .attribution cite{display:block;letter-spacing:-.025em;color:rgba(0,0,0,.6)}
.quoteblock.abstract blockquote::before,.quoteblock.excerpt blockquote::before,.quoteblock .quoteblock blockquote::before{display:none}
.quoteblock.abstract blockquote,.quoteblock.abstract p,.quoteblock.excerpt blockquote,.quoteblock.excerpt p,.quoteblock .quoteblock blockquote,.quoteblock .quoteblock p{line-height:1.6;word-spacing:0}
.quoteblock.abstract{margin:0 1em 1.25em;display:block}
.quoteblock.abstract>.title{margin:0 0 .375em;font-size:1.15em;text-align:center}
.quoteblock.excerpt,.quoteblock .quoteblock{margin:0 0 1.25em;padding:0 0 .25em 1em;border-left:.25em solid #dddddf}
.quoteblock.excerpt blockquote,.quoteblock.excerpt p,.quoteblock .quoteblock blockquote,.quoteblock .quoteblock p{color:inherit;font-size:1.0625rem}
.quoteblock.excerpt .attribution,.quoteblock .quoteblock .attribution{color:inherit;text-align:left;margin-right:0}
table.tableblock{max-width:100%;border-collapse:separate}
p.tableblock:last-child{margin-bottom:0}
td.tableblock>.content{margin-bottom:-1.25em}
table.tableblock,th.tableblock,td.tableblock{border:0 solid #dedede}
table.grid-all>thead>tr>.tableblock,table.grid-all>tbody>tr>.tableblock{border-width:0 1px 1px 0}
table.grid-all>tfoot>tr>.tableblock{border-width:1px 1px 0 0}
table.grid-cols>*>tr>.tableblock{border-width:0 1px 0 0}
table.grid-rows>thead>tr>.tableblock,table.grid-rows>tbody>tr>.tableblock{border-width:0 0 1px}
table.grid-rows>tfoot>tr>.tableblock{border-width:1px 0 0}
table.grid-all>*>tr>.tableblock:last-child,table.grid-cols>*>tr>.tableblock:last-child{border-right-width:0}
table.grid-all>tbody>tr:last-child>.tableblock,table.grid-all>thead:last-child>tr>.tableblock,table.grid-rows>tbody>tr:last-child>.tableblock,table.grid-rows>thead:last-child>tr>.tableblock{border-bottom-width:0}
table.frame-all{border-width:1px}
table.frame-sides{border-width:0 1px}
table.frame-topbot,table.frame-ends{border-width:1px 0}
table.stripes-all tr,table.stripes-odd tr:nth-of-type(odd){background:#f8f8f7}
table.stripes-none tr,table.stripes-odd tr:nth-of-type(even){background:none}
th.halign-left,td.halign-left{text-align:left}
th.halign-right,td.halign-right{text-align:right}
th.halign-center,td.halign-center{text-align:center}
th.valign-top,td.valign-top{vertical-align:top}
th.valign-bottom,td.valign-bottom{vertical-align:bottom}
th.valign-middle,td.valign-middle{vertical-align:middle}
table thead th,table tfoot th{font-weight:bold}
tbody tr th{display:table-cell;line-height:1.6;background:#f7f8f7}
tbody tr th,tbody tr th p,tfoot tr th,tfoot tr th p{color:rgba(0,0,0,.8);font-weight:bold}
p.tableblock>code:only-child{background:none;padding:0}
p.tableblock{font-size:1em}
td>div.verse{white-space:pre}
ol{margin-left:1.75em}
ul li ol{margin-left:1.5em}
dl dd{margin-left:1.125em}
dl dd:last-child,dl dd:last-child>:last-child{margin-bottom:0}
ol>li p,ul>li p,ul dd,ol dd,.olist .olist,.ulist .ulist,.ulist .olist,.olist .ulist{margin-bottom:.625em}
ul.checklist,ul.none,ol.none,ul.no-bullet,ol.no-bullet,ol.unnumbered,ul.unstyled,ol.unstyled{list-style-type:none}
ul.no-bullet,ol.no-bullet,ol.unnumbered{margin-left:.625em}
ul.unstyled,ol.unstyled{margin-left:0}
ul.checklist{margin-left:.625em}
ul.checklist li>p:first-child>.fa-square-o:first-child,ul.checklist li>p:first-child>.fa-check-square-o:first-child{width:1.25em;font-size:.8em;position:relative;bottom:.125em}
ul.checklist li>p:first-child>input[type="checkbox"]:first-child{margin-right:.25em}
ul.inline{display:-ms-flexbox;display:-webkit-box;display:flex;-ms-flex-flow:row wrap;-webkit-flex-flow:row wrap;flex-flow:row wrap;list-style:none;margin:0 0 .625em -1.25em}
ul.inline>li{margin-left:1.25em}
.unstyled dl dt{font-weight:400;font-style:normal}
ol.arabic{list-style-type:decimal}
ol.decimal{list-style-type:decimal-leading-zero}
ol.loweralpha{list-style-type:lower-alpha}
ol.upperalpha{list-style-type:upper-alpha}
ol.lowerroman{list-style-type:lower-roman}
ol.upperroman{list-style-type:upper-roman}
ol.lowergreek{list-style-type:lower-greek}
.hdlist>table,.colist>table{border:0;background:none}
.hdlist>table>tbody>tr,.colist>table>tbody>tr{background:none}
td.hdlist1,td.hdlist2{vertical-align:top;padding:0 .625em}
td.hdlist1{font-weight:bold;padding-bottom:1.25em}
.literalblock+.colist,.listingblock+.colist{margin-top:-.5em}
.colist td:not([class]):first-child{padding:.4em .75em 0;line-height:1;vertical-align:top}
.colist td:not([class]):first-child img{max-width:none}
.colist td:not([class]):last-child{padding:.25em 0}
.thumb,.th{line-height:0;display:inline-block;border:solid 4px #fff;-webkit-box-shadow:0 0 0 1px #ddd;box-shadow:0 0 0 1px #ddd}
.imageblock.left{margin:.25em .625em 1.25em 0}
.imageblock.right{margin:.25em 0 1.25em .625em}
.imageblock>.title{margin-bottom:0}
.imageblock.thumb,.imageblock.th{border-width:6px}
.imageblock.thumb>.title,.imageblock.th>.title{padding:0 .125em}
.image.left,.image.right{margin-top:.25em;margin-bottom:.25em;display:inline-block;line-height:0}
.image.left{margin-right:.625em}
.image.right{margin-left:.625em}
a.image{text-decoration:none;display:inline-block}
a.image object{pointer-events:none}
sup.footnote,sup.footnoteref{font-size:.875em;position:static;vertical-align:super}
sup.footnote a,sup.footnoteref a{text-decoration:none}
sup.footnote a:active,sup.footnoteref a:active{text-decoration:underline}
#footnotes{padding-top:.75em;padding-bottom:.75em;margin-bottom:.625em}
#footnotes hr{width:20%;min-width:6.25em;margin:-.25em 0 .75em;border-width:1px 0 0}
#footnotes .footnote{padding:0 .375em 0 .225em;line-height:1.3334;font-size:.875em;margin-left:1.2em;margin-bottom:.2em}
#footnotes .footnote a:first-of-type{font-weight:bold;text-decoration:none;margin-left:-1.05em}
#footnotes .footnote:last-of-type{margin-bottom:0}
#content #footnotes{margin-top:-.625em;margin-bottom:0;padding:.75em 0}
.gist .file-data>table{border:0;background:#fff;width:100%;margin-bottom:0}
.gist .file-data>table td.line-data{width:99%}
div.unbreakable{page-break-inside:avoid}
.big{font-size:larger}
.small{font-size:smaller}
.underline{text-decoration:underline}
.overline{text-decoration:overline}
.line-through{text-decoration:line-through}
.aqua{color:#00bfbf}
.aqua-background{background-color:#00fafa}
.black{color:#000}
.black-background{background-color:#000}
.blue{color:#0000bf}
.blue-background{background-color:#0000fa}
.fuchsia{color:#bf00bf}
.fuchsia-background{background-color:#fa00fa}
.gray{color:#606060}
.gray-background{background-color:#7d7d7d}
.green{color:#006000}
.green-background{background-color:#007d00}
.lime{color:#00bf00}
.lime-background{background-color:#00fa00}
.maroon{color:#600000}
.maroon-background{background-color:#7d0000}
.navy{color:#000060}
.navy-background{background-color:#00007d}
.olive{color:#606000}
.olive-background{background-color:#7d7d00}
.purple{color:#600060}
.purple-background{background-color:#7d007d}
.red{color:#bf0000}
.red-background{background-color:#fa0000}
.silver{color:#909090}
.silver-background{background-color:#bcbcbc}
.teal{color:#006060}
.teal-background{background-color:#007d7d}
.white{color:#bfbfbf}
.white-background{background-color:#fafafa}
.yellow{color:#bfbf00}
.yellow-background{background-color:#fafa00}
span.icon>.fa{cursor:default}
a span.icon>.fa{cursor:inherit}
.admonitionblock td.icon [class^="fa icon-"]{font-size:2.5em;text-shadow:1px 1px 2px rgba(0,0,0,.5);cursor:default}
.admonitionblock td.icon .icon-note::before{content:"\f05a";color:#19407c}
.admonitionblock td.icon .icon-tip::before{content:"\f0eb";text-shadow:1px 1px 2px rgba(155,155,0,.8);color:#111}
.admonitionblock td.icon .icon-warning::before{content:"\f071";color:#bf6900}
.admonitionblock td.icon .icon-caution::before{content:"\f06d";color:#bf3400}
.admonitionblock td.icon .icon-important::before{content:"\f06a";color:#bf0000}
.conum[data-value]{display:inline-block;color:#fff!important;background-color:rgba(0,0,0,.8);-webkit-border-radius:100px;border-radius:100px;text-align:center;font-size:.75em;width:1.67em;height:1.67em;line-height:1.67em;font-family:"Open Sans","DejaVu Sans",sans-serif;font-style:normal;font-weight:bold}
.conum[data-value] *{color:#fff!important}
.conum[data-value]+b{display:none}
.conum[data-value]::after{content:attr(data-value)}
pre .conum[data-value]{position:relative;top:-.125em}
b.conum *{color:inherit!important}
.conum:not([data-value]):empty{display:none}
dt,th.tableblock,td.content,div.footnote{text-rendering:optimizeLegibility}
h1,h2,p,td.content,span.alt{letter-spacing:-.01em}
p strong,td.content strong,div.footnote strong{letter-spacing:-.005em}
p,blockquote,dt,td.content,span.alt{font-size:1.0625rem}
p{margin-bottom:1.25rem}
.sidebarblock p,.sidebarblock dt,.sidebarblock td.content,p.tableblock{font-size:1em}
.exampleblock>.content{background-color:#fffef7;border-color:#e0e0dc;-webkit-box-shadow:0 1px 4px #e0e0dc;box-shadow:0 1px 4px #e0e0dc}
.print-only{display:none!important}
@page{margin:1.25cm .75cm}
@media print{*{-webkit-box-shadow:none!important;box-shadow:none!important;text-shadow:none!important}
html{font-size:80%}
a{color:inherit!important;text-decoration:underline!important}
a.bare,a[href^="#"],a[href^="mailto:"]{text-decoration:none!important}
a[href^="http:"]:not(.bare)::after,a[href^="https:"]:not(.bare)::after{content:"(" attr(href) ")";display:inline-block;font-size:.875em;padding-left:.25em}
abbr[title]::after{content:" (" attr(title) ")"}
pre,blockquote,tr,img,object,svg{page-break-inside:avoid}
thead{display:table-header-group}
svg{max-width:100%}
p,blockquote,dt,td.content{font-size:1em;orphans:3;widows:3}
h2,h3,#toctitle,.sidebarblock>.content>.title{page-break-after:avoid}
#toc,.sidebarblock,.exampleblock>.content{background:none!important}
#toc{border-bottom:1px solid #dddddf!important;padding-bottom:0!important}
body.book #header{text-align:center}
body.book #header>h1:first-child{border:0!important;margin:2.5em 0 1em}
body.book #header .details{border:0!important;display:block;padding:0!important}
body.book #header .details span:first-child{margin-left:0!important}
body.book #header .details br{display:block}
body.book #header .details br+span::before{content:none!important}
body.book #toc{border:0!important;text-align:left!important;padding:0!important;margin:0!important}
body.book #toc,body.book #preamble,body.book h1.sect0,body.book .sect1>h2{page-break-before:always}
.listingblock code[data-lang]::before{display:block}
#footer{padding:0 .9375em}
.hide-on-print{display:none!important}
.print-only{display:block!important}
.hide-for-print{display:none!important}
.show-for-print{display:inherit!important}}
@media print,amzn-kf8{#header>h1:first-child{margin-top:1.25rem}
.sect1{padding:0!important}
.sect1+.sect1{border:0}
#footer{background:none}
#footer-text{color:rgba(0,0,0,.6);font-size:.9em}}
@media amzn-kf8{#header,#content,#footnotes,#footer{padding:0}}

      </style>
      <link href='https://fonts.googleapis.com/css?family=Noto+Serif' rel='stylesheet' type='text/css'>
      <link href='https://fonts.googleapis.com/css?family=Open+Sans:400,300,300italic,400italic,600,600italic,700,700italic,800,800italic' rel='stylesheet' type='text/css'>
      <link href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/4.7.0/css/font-awesome.min.css" rel="stylesheet">
      <link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/styles/default.min.css">
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/highlight.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/asciidoc.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/yaml.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/dockerfile.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/makefile.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/go.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/rust.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/haskell.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/typescript.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/scss.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/less.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/handlebars.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/groovy.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/scala.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/bash.min.js"></script>
      <script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/9.9.0/languages/ini.min.js"></script>
      <script>hljs.initHighlightingOnLoad();</script>
    </head>
    <body>
      <div id="wrapper">
        <div class="article">
          <h1>Spark core Insight</h1>
<div id="preamble">
<div class="sectionbody">
<div class="exampleblock">
<div class="title">目标</div>
<div class="content">
<div class="olist arabic">
<ol class="arabic">
<li>
<p>深入理解 RDD 的内在逻辑</p>
</li>
<li>
<p>能够使用 RDD 的算子</p>
</li>
<li>
<p>理解 RDD 算子的 Shuffle 和缓存</p>
</li>
<li>
<p>理解 RDD 整体的使用流程</p>
</li>
<li>
<p>理解 RDD 的调度原理</p>
</li>
<li>
<p>理解 Spark 中常见的分布式变量共享方式</p>
</li>
</ol>
</div>
</div>
</div>
</div>
<div id="toc" class="toc">
<div id="toctitle">Table of Contents</div>
<ul class="sectlevel1">
<li><a href="#_1_深入_rdd">1. 深入 RDD</a>
<ul class="sectlevel2">
<li><a href="#_1_1_案例">1.1. 案例</a></li>
<li><a href="#_1_2_再谈_rdd">1.2. 再谈 RDD</a></li>
</ul>
</li>
<li><a href="#_2_rdd_的算子">2. RDD 的算子</a>
<ul class="sectlevel2">
<li><a href="#_2_1_transformations_算子">2.1. Transformations 算子</a></li>
<li><a href="#_2_2_action_算子">2.2. Action 算子</a></li>
<li><a href="#_2_3_rdd_对不同类型数据的支持">2.3. RDD 对不同类型数据的支持</a></li>
</ul>
</li>
<li><a href="#_3_rdd_的_shuffle_和分区">3. RDD 的 Shuffle 和分区</a>
<ul class="sectlevel2">
<li><a href="#_3_1_rdd_的分区操作">3.1. RDD 的分区操作</a></li>
<li><a href="#_3_2_rdd_的_shuffle_是什么">3.2. RDD 的 Shuffle 是什么</a></li>
<li><a href="#_3_3_rdd_的_shuffle_原理">3.3. RDD 的 Shuffle 原理</a></li>
</ul>
</li>
<li><a href="#_4_缓存">4. 缓存</a>
<ul class="sectlevel2">
<li><a href="#_4_1_缓存的意义">4.1. 缓存的意义</a></li>
<li><a href="#_4_2_缓存相关的_api">4.2. 缓存相关的 API</a></li>
<li><a href="#_4_3_缓存级别">4.3. 缓存级别</a></li>
</ul>
</li>
<li><a href="#_5_checkpoint">5. Checkpoint</a>
<ul class="sectlevel2">
<li><a href="#_5_1_checkpoint_的作用">5.1. Checkpoint 的作用</a></li>
<li><a href="#_5_2_使用_checkpoint">5.2. 使用 Checkpoint</a></li>
</ul>
</li>
</ul>
</div>
</div>
<div class="sect1">
<h2 id="_1_深入_rdd">1. 深入 RDD</h2>
<div class="sectionbody">
<div class="exampleblock">
<div class="title">目标</div>
<div class="content">
<div class="olist arabic">
<ol class="arabic">
<li>
<p>深入理解 RDD 的内在逻辑, 以及 RDD 的内部属性(RDD 由什么组成)</p>
</li>
</ol>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_1_1_案例">1.1. 案例</h3>
<div class="exampleblock">
<div class="title">需求</div>
<div class="content">
<div class="ulist">
<ul>
<li>
<p>给定一个网站的访问记录, 俗称 Access log</p>
</li>
<li>
<p>计算其中出现的独立 IP, 以及其访问的次数</p>
</li>
</ul>
</div>
</div>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val config = new SparkConf().setAppName("ip_ana").setMaster("local[6]")
val sc = new SparkContext(config)

val result = sc.textFile("dataset/access_log_sample.txt")
  .map(item =&gt; (item.split(" ")(0), 1))
  .filter(item =&gt; StringUtils.isNotBlank(item._1))
  .reduceByKey((curr, agg) =&gt; curr + agg)
  .sortBy(item =&gt; item._2, false)
  .take(10)

result.foreach(item =&gt; println(item))</code></pre>
</div>
</div>
<div class="paragraph">
<p>针对这个小案例, 我们问出互相关联但是又方向不同的五个问题</p>
</div>
<div class="qlist qanda">
<ol>
<li>
<p><em>假设要针对整个网站的历史数据进行处理, 量有 1T, 如何处理?</em></p>
<div class="sidebarblock">
<div class="content">
<div class="paragraph">
<p>放在集群中, 利用集群多台计算机来并行处理</p>
</div>
</div>
</div>
</li>
<li>
<p><em>如何放在集群中运行?</em></p>
<div class="sidebarblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/6088be299490adbaaeece8717ae985e8.png" alt="6088be299490adbaaeece8717ae985e8" width="800">
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">简单来讲, 并行计算就是同时使用多个计算资源解决一个问题, 有如下四个要点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>要解决的问题必须可以分解为多个可以并发计算的部分</p>
</li>
<li>
<p>每个部分要可以在不同处理器上被同时执行</p>
</li>
<li>
<p><strong>需要一个共享内存的机制</strong></p>
</li>
<li>
<p>需要一个总体上的协作机制来进行调度</p>
</li>
</ul>
</div>
</dd>
</dl>
</div>
</div>
</div>
</li>
<li>
<p><em>如果放在集群中的话, 可能要对整个计算任务进行分解, 如何分解?</em></p>
<div class="sidebarblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/f738dbe3df690bc0ba8f580a3e2d1112.png" alt="f738dbe3df690bc0ba8f580a3e2d1112" width="800">
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">概述</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>对于 HDFS 中的文件, 是分为不同的 Block 的</p>
</li>
<li>
<p>在进行计算的时候, 就可以按照 Block 来划分, 每一个 Block 对应一个不同的计算单元</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">扩展</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>RDD</code> 并没有真实的存放数据, 数据是从 HDFS 中读取的, 在计算的过程中读取即可</p>
</li>
<li>
<p><code>RDD</code> 至少是需要可以 <strong>分片</strong> 的, 因为HDFS中的文件就是分片的, <code>RDD</code> 分片的意义在于表示对源数据集每个分片的计算, <code>RDD</code> 可以分片也意味着 <strong>可以并行计算</strong></p>
</li>
</ul>
</div>
</dd>
</dl>
</div>
</div>
</div>
</li>
<li>
<p><em>移动数据不如移动计算是一个基础的优化, 如何做到?</em></p>
<div class="sidebarblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/1d344ab200bd12866c26ca2ea6ab1e37.png" alt="1d344ab200bd12866c26ca2ea6ab1e37" width="800">
</div>
</div>
<div class="paragraph">
<p>每一个计算单元需要记录其存储单元的位置, 尽量调度过去</p>
</div>
</div>
</div>
</li>
<li>
<p><em>在集群中运行, 需要很多节点之间配合, 出错的概率也更高, 出错了怎么办?</em></p>
<div class="sidebarblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/5c7bef41f177a96e99c7ad8a500b7310.png" alt="5c7bef41f177a96e99c7ad8a500b7310" width="800">
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">RDD1 &#8594; RDD2 &#8594; RDD3 这个过程中, RDD2 出错了, 有两种办法可以解决</dt>
<dd>
<div class="olist arabic">
<ol class="arabic">
<li>
<p>缓存 RDD2 的数据, 直接恢复 RDD2, 类似 HDFS 的备份机制</p>
</li>
<li>
<p>记录 RDD2 的依赖关系, 通过其父级的 RDD 来恢复 RDD2, 这种方式会少很多数据的交互和保存</p>
</li>
</ol>
</div>
</dd>
<dt class="hdlist1">如何通过父级 RDD 来恢复?</dt>
<dd>
<div class="olist arabic">
<ol class="arabic">
<li>
<p>记录 RDD2 的父亲是 RDD1</p>
</li>
<li>
<p>记录 RDD2 的计算函数, 例如记录 <code>RDD2 = RDD1.map(&#8230;&#8203;)</code>, <code>map(&#8230;&#8203;)</code> 就是计算函数</p>
</li>
<li>
<p>当 RDD2 计算出错的时候, 可以通过父级 RDD 和计算函数来恢复 RDD2</p>
</li>
</ol>
</div>
</dd>
</dl>
</div>
</div>
</div>
</li>
<li>
<p><em>假如任务特别复杂, 流程特别长, 有很多 RDD 之间有依赖关系, 如何优化?</em></p>
<div class="sidebarblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/dc87ed7f9b653bccb43d099bbb4f537f.png" alt="dc87ed7f9b653bccb43d099bbb4f537f" width="800">
</div>
</div>
<div class="paragraph">
<p>上面提到了可以使用依赖关系来进行容错, 但是如果依赖关系特别长的时候, 这种方式其实也比较低效, 这个时候就应该使用另外一种方式, 也就是记录数据集的状态</p>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">在 Spark 中有两个手段可以做到</dt>
<dd>
<div class="olist arabic">
<ol class="arabic">
<li>
<p>缓存</p>
</li>
<li>
<p>Checkpoint</p>
</li>
</ol>
</div>
</dd>
</dl>
</div>
</div>
</div>
</li>
</ol>
</div>
</div>
<div class="sect2">
<h3 id="_1_2_再谈_rdd">1.2. 再谈 RDD</h3>
<div class="exampleblock">
<div class="title">目标</div>
<div class="content">
<div class="olist arabic">
<ol class="arabic">
<li>
<p>理解 RDD 为什么会出现</p>
</li>
<li>
<p>理解 RDD 的主要特点</p>
</li>
<li>
<p>理解 RDD 的五大属性</p>
</li>
</ol>
</div>
</div>
</div>
<div class="sect3">
<h4 id="_1_2_1_rdd_为什么会出现">1.2.1. RDD 为什么会出现?</h4>
<div class="dlist">
<dl>
<dt class="hdlist1">在 RDD 出现之前, 当时 MapReduce 是比较主流的, 而 MapReduce 如何执行迭代计算的任务呢?</dt>
<dd>
<div class="openblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/306061ee343d8515ecafbce43bc54bc6.png" alt="306061ee343d8515ecafbce43bc54bc6" width="800">
</div>
</div>
<div class="paragraph">
<p>多个 MapReduce 任务之间没有基于内存的数据共享方式, 只能通过磁盘来进行共享</p>
</div>
<div class="paragraph">
<p>这种方式明显比较低效</p>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">RDD 如何解决迭代计算非常低效的问题呢?</dt>
<dd>
<div class="openblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/4fc644616fb13ef896eb3a8cea5d3bd7.png" alt="4fc644616fb13ef896eb3a8cea5d3bd7" width="800">
</div>
</div>
<div class="paragraph">
<p>在 Spark 中, 其实最终 Job3 从逻辑上的计算过程是: <code>Job3 = (Job1.map).filter</code>, 整个过程是共享内存的, 而不需要将中间结果存放在可靠的分布式文件系统中</p>
</div>
<div class="paragraph">
<p>这种方式可以在保证容错的前提下, 提供更多的灵活, 更快的执行速度, RDD 在执行迭代型任务时候的表现可以通过下面代码体现</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-java hljs" data-lang="java">// 线性回归
val points = sc.textFile(...)
	.map(...)
	.persist(...)
val w = randomValue
for (i &lt;- 1 to 10000) {
    val gradient = points.map(p =&gt; p.x * (1 / (1 + exp(-p.y * (w dot p.x))) - 1) * p.y)
    	.reduce(_ + _)
    w -= gradient
}</code></pre>
</div>
</div>
<div class="paragraph">
<p>在这个例子中, 进行了大致 10000 次迭代, 如果在 MapReduce 中实现, 可能需要运行很多 Job, 每个 Job 之间都要通过 HDFS 共享结果, 熟快熟慢一窥便知</p>
</div>
</div>
</div>
</dd>
</dl>
</div>
</div>
<div class="sect3">
<h4 id="_1_2_2_rdd_的特点">1.2.2. RDD 的特点</h4>
<div class="dlist">
<dl>
<dt class="hdlist1">RDD 不仅是数据集, 也是编程模型</dt>
<dd>
<div class="openblock">
<div class="content">
<div class="paragraph">
<p>RDD 即是一种数据结构, 同时也提供了上层 API, 同时 RDD 的 API 和 Scala 中对集合运算的 API 非常类似, 同样也都是各种算子</p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/02adfc1bcd91e70c1619fc6a67b13f92.png" alt="02adfc1bcd91e70c1619fc6a67b13f92" width="800">
</div>
</div>
<div class="paragraph">
<p>RDD 的算子大致分为两类:</p>
</div>
<div class="ulist">
<ul>
<li>
<p>Transformation 转换操作, 例如 <code>map</code> <code>flatMap</code> <code>filter</code> 等</p>
</li>
<li>
<p>Action 动作操作, 例如 <code>reduce</code> <code>collect</code> <code>show</code> 等</p>
</li>
</ul>
</div>
<div class="paragraph">
<p>执行 RDD 的时候, 在执行到转换操作的时候, 并不会立刻执行, 直到遇见了 Action 操作, 才会触发真正的执行, 这个特点叫做 <strong>惰性求值</strong></p>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">RDD 可以分区</dt>
<dd>
<div class="openblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/2ba2cc9ad8e745c26df482b4e968c802.png" alt="2ba2cc9ad8e745c26df482b4e968c802" width="800">
</div>
</div>
<div class="paragraph">
<p>RDD 是一个分布式计算框架, 所以, 一定是要能够进行分区计算的, 只有分区了, 才能利用集群的并行计算能力</p>
</div>
<div class="paragraph">
<p>同时, RDD 不需要始终被具体化, 也就是说: RDD 中可以没有数据, 只要有足够的信息知道自己是从谁计算得来的就可以, 这是一种非常高效的容错方式</p>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">RDD 是只读的</dt>
<dd>
<div class="openblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ed6a534cfe0a56de3c34ac6e1e8d504e.png" alt="ed6a534cfe0a56de3c34ac6e1e8d504e" width="800">
</div>
</div>
<div class="paragraph">
<p>RDD 是只读的, 不允许任何形式的修改. 虽说不能因为 RDD 和 HDFS 是只读的, 就认为分布式存储系统必须设计为只读的. 但是设计为只读的, 会显著降低问题的复杂度, 因为 RDD 需要可以容错, 可以惰性求值, 可以移动计算, 所以很难支持修改.</p>
</div>
<div class="ulist">
<ul>
<li>
<p>RDD2 中可能没有数据, 只是保留了依赖关系和计算函数, 那修改啥?</p>
</li>
<li>
<p>如果因为支持修改, 而必须保存数据的话, 怎么容错?</p>
</li>
<li>
<p>如果允许修改, 如何定位要修改的那一行? RDD 的转换是粗粒度的, 也就是说, RDD 并不感知具体每一行在哪.</p>
</li>
</ul>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">RDD 是可以容错的</dt>
<dd>
<div class="openblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/5c7bef41f177a96e99c7ad8a500b7310.png" alt="5c7bef41f177a96e99c7ad8a500b7310" width="800">
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">RDD 的容错有两种方式</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>保存 RDD 之间的依赖关系, 以及计算函数, 出现错误重新计算</p>
</li>
<li>
<p>直接将 RDD 的数据存放在外部存储系统, 出现错误直接读取, Checkpoint</p>
</li>
</ul>
</div>
</dd>
</dl>
</div>
</div>
</div>
</dd>
</dl>
</div>
</div>
<div class="sect3">
<h4 id="_1_2_3_什么叫做弹性分布式数据集">1.2.3. 什么叫做弹性分布式数据集</h4>
<div class="dlist">
<dl>
<dt class="hdlist1">分布式</dt>
<dd>
<p>RDD 支持分区, 可以运行在集群中</p>
</dd>
<dt class="hdlist1">弹性</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>RDD 支持高效的容错</p>
</li>
<li>
<p>RDD 中的数据即可以缓存在内存中, 也可以缓存在磁盘中, 也可以缓存在外部存储中</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">数据集</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>RDD 可以不保存具体数据, 只保留创建自己的必备信息, 例如依赖和计算函数</p>
</li>
<li>
<p>RDD 也可以缓存起来, 相当于存储具体数据</p>
</li>
</ul>
</div>
</dd>
</dl>
</div>
<div class="sidebarblock">
<div class="content">
<div class="title">总结: RDD 的五大属性</div>
<div class="paragraph">
<p>首先整理一下上面所提到的 RDD 所要实现的功能:</p>
</div>
<div class="olist arabic">
<ol class="arabic">
<li>
<p>RDD 有分区</p>
</li>
<li>
<p>RDD 要可以通过依赖关系和计算函数进行容错</p>
</li>
<li>
<p>RDD 要针对数据本地性进行优化</p>
</li>
<li>
<p>RDD 支持 MapReduce 形式的计算, 所以要能够对数据进行 Shuffled</p>
</li>
</ol>
</div>
<div class="paragraph">
<p>对于 RDD 来说, 其中应该有什么内容呢? 如果站在 RDD 设计者的角度上, 这个类中, 至少需要什么属性?</p>
</div>
<div class="ulist">
<ul>
<li>
<p><code>Partition List</code> 分片列表, 记录 RDD 的分片, 可以在创建 RDD 的时候指定分区数目, 也可以通过算子来生成新的 RDD 从而改变分区数目</p>
</li>
<li>
<p><code>Compute Function</code> 为了实现容错, 需要记录 RDD 之间转换所执行的计算函数</p>
</li>
<li>
<p><code>RDD Dependencies</code> RDD 之间的依赖关系, 要在 RDD 中记录其上级 RDD 是谁, 从而实现容错和计算</p>
</li>
<li>
<p><code>Partitioner</code> 为了执行 Shuffled 操作, 必须要有一个函数用来计算数据应该发往哪个分区</p>
</li>
<li>
<p><code>Preferred Location</code> 优先位置, 为了实现数据本地性操作, 从而移动计算而不是移动存储, 需要记录每个 RDD 分区最好应该放置在什么位置</p>
</li>
</ul>
</div>
</div>
</div>
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="_2_rdd_的算子">2. RDD 的算子</h2>
<div class="sectionbody">
<div class="exampleblock">
<div class="title">目标</div>
<div class="content">
<div class="olist arabic">
<ol class="arabic">
<li>
<p>理解 RDD 的算子分类, 以及其特性</p>
</li>
<li>
<p>理解常见算子的使用</p>
</li>
</ol>
</div>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">分类</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="paragraph">
<p>RDD 中的算子从功能上分为两大类</p>
</div>
<div class="olist arabic">
<ol class="arabic">
<li>
<p>Transformation(转换) 它会在一个已经存在的 RDD 上创建一个新的 RDD, 将旧的 RDD 的数据转换为另外一种形式后放入新的 RDD</p>
</li>
<li>
<p>Action(动作) 执行各个分区的计算任务, 将的到的结果返回到 Driver 中</p>
</li>
</ol>
</div>
<div class="paragraph">
<p>RDD 中可以存放各种类型的数据, 那么对于不同类型的数据, RDD 又可以分为三类</p>
</div>
<div class="ulist">
<ul>
<li>
<p>针对基础类型(例如 String)处理的普通算子</p>
</li>
<li>
<p>针对 <code>Key-Value</code> 数据处理的 <code>byKey</code> 算子</p>
</li>
<li>
<p>针对数字类型数据处理的计算算子</p>
</li>
</ul>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">特点</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="ulist">
<ul>
<li>
<p>Spark 中所有的 Transformations 是 Lazy(惰性) 的, 它们不会立即执行获得结果. 相反, 它们只会记录在数据集上要应用的操作. 只有当需要返回结果给 Driver 时, 才会执行这些操作, 通过 DAGScheduler 和 TaskScheduler 分发到集群中运行, 这个特性叫做 <strong>惰性求值</strong></p>
</li>
<li>
<p>默认情况下, 每一个 Action 运行的时候, 其所关联的所有 Transformation RDD 都会重新计算, 但是也可以使用 <code>presist</code> 方法将 RDD 持久化到磁盘或者内存中. 这个时候为了下次可以更快的访问, 会把数据保存到集群上.</p>
</li>
</ul>
</div>
</div>
</div>
</dd>
</dl>
</div>
<div class="sect2">
<h3 id="_2_1_transformations_算子">2.1. Transformations 算子</h3>
<table class="tableblock frame-all grid-all stretch">
<colgroup>
<col style="width: 8%;">
<col>
</colgroup>
<tbody>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock">Transformation function</p></th>
<td class="tableblock halign-left valign-top"><p class="tableblock">解释</p></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>map(T &#8658; U)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">sc.parallelize(Seq(1, 2, 3))
  .map( num =&gt; num * 10 )
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/57c2f77284bfa8f99ade091fdd7e9f83.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/57c2f77284bfa8f99ade091fdd7e9f83.png" alt="57c2f77284bfa8f99ade091fdd7e9f83" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/c59d44296918b864a975ebbeb60d4c04.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/c59d44296918b864a975ebbeb60d4c04.png" alt="c59d44296918b864a975ebbeb60d4c04" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>把 RDD 中的数据 一对一 的转为另一种形式</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">签名</dt>
<dd>
<p><code>def map[U: ClassTag](f: T &#8658; U): RDD[U]</code></p>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>f</code> &#8594; Map 算子是 <code>原RDD &#8594; 新RDD</code> 的过程, 传入函数的参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>Map 是一对一, 如果函数是 <code>String &#8594; Array[String]</code> 则新的 RDD 中每条数据就是一个数组</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>flatMap(T &#8658; List[U])</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">sc.parallelize(Seq("Hello lily", "Hello lucy", "Hello tim"))
  .flatMap( line =&gt; line.split(" ") )
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ec39594f30ca4d59e2ef5cdc60387866.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ec39594f30ca4d59e2ef5cdc60387866.png" alt="ec39594f30ca4d59e2ef5cdc60387866" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/f6c4feba14bb71372aa0cb678067c6a8.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/f6c4feba14bb71372aa0cb678067c6a8.png" alt="f6c4feba14bb71372aa0cb678067c6a8" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>FlatMap 算子和 Map 算子类似, 但是 FlatMap 是一对多</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<p><code>def flatMap[U: ClassTag](f: T &#8658; List[U]): RDD[U]</code></p>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>f</code> &#8594; 参数是原 RDD 数据, 返回值是经过函数转换的新 RDD 的数据, 需要注意的是返回值是一个集合, 集合中的数据会被展平后再放入新的 RDD</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>flatMap 其实是两个操作, 是 <code>map + flatten</code>, 也就是先转换, 后把转换而来的 List 展开</p>
</li>
<li>
<p>Spark 中并没有直接展平 RDD 中数组的算子, 可以使用 <code>flatMap</code> 做这件事</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>filter(T &#8658; Boolean)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">sc.parallelize(Seq(1, 2, 3))
  .filter( value =&gt; value &gt;= 3 )
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/25a7aef5e2b8a39145d503f4652cc945.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/25a7aef5e2b8a39145d503f4652cc945.png" alt="25a7aef5e2b8a39145d503f4652cc945" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/05cdb79abd41a7b5baa41a4c62870d73.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/05cdb79abd41a7b5baa41a4c62870d73.png" alt="05cdb79abd41a7b5baa41a4c62870d73" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>Filter</code> 算子的主要作用是过滤掉不需要的内容</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>mapPartitions(List[T] &#8658; List[U])</code></p></th>
<td class="tableblock halign-left valign-top"><p class="tableblock"><strong>RDD[T] &#8658; RDD[U]</strong> 和 map 类似, 但是针对整个分区的数据转换</p></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>mapPartitionsWithIndex</code></p></th>
<td class="tableblock halign-left valign-top"><p class="tableblock">和 mapPartitions 类似, 只是在函数中增加了分区的 Index</p></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>mapValues</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">sc.parallelize(Seq(("a", 1), ("b", 2), ("c", 3)))
  .mapValues( value =&gt; value * 10 )
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/7a8b280a054fdab8e8d14549f67b85f9.png" alt="7a8b280a054fdab8e8d14549f67b85f9" width="linkhttps://doc-1256053707.cos.ap-beijing.myqcloud.com/7a8b280a054fdab8e8d14549f67b85f9.png">
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/5551847febe453b134f3a4009df01bec.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/5551847febe453b134f3a4009df01bec.png" alt="5551847febe453b134f3a4009df01bec" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>MapValues 只能作用于 Key-Value 型数据, 和 Map 类似, 也是使用函数按照转换数据, 不同点是 MapValues 只转换 Key-Value 中的 Value</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>sample(withReplacement, fraction, seed)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">sc.parallelize(Seq(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
  .sample(withReplacement = true, 0.6, 2)
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/03139edb0211652195dccea955f3a9b3.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/03139edb0211652195dccea955f3a9b3.png" alt="03139edb0211652195dccea955f3a9b3" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ccd1ae121f6f6852158c044441437f04.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ccd1ae121f6f6852158c044441437f04.png" alt="ccd1ae121f6f6852158c044441437f04" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>Sample 算子可以从一个数据集中抽样出来一部分, 常用作于减小数据集以保证运行速度, 并且尽可能少规律的损失</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>Sample 接受第一个参数为`withReplacement`, 意为是否取样以后是否还放回原数据集供下次使用, 简单的说, 如果这个参数的值为 true, 则抽样出来的数据集中可能会有重复</p>
</li>
<li>
<p>Sample 接受第二个参数为`fraction`, 意为抽样的比例</p>
</li>
<li>
<p>Sample 接受第三个参数为`seed`, 随机数种子, 用于 Sample 内部随机生成下标, 一般不指定, 使用默认值</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>union(other)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd1 = sc.parallelize(Seq(1, 2, 3))
val rdd2 = sc.parallelize(Seq(4, 5, 6))
rdd1.union(rdd2)
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/5f31c2c44aa66db3027fea4624a3c4eb.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/5f31c2c44aa66db3027fea4624a3c4eb.png" alt="5f31c2c44aa66db3027fea4624a3c4eb" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/2a8b7d10930251ae32d6d276ab7f41f8.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/2a8b7d10930251ae32d6d276ab7f41f8.png" alt="2a8b7d10930251ae32d6d276ab7f41f8" width="800"></a>
</div>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>intersection(other)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd1 = sc.parallelize(Seq(1, 2, 3, 4, 5))
val rdd2 = sc.parallelize(Seq(4, 5, 6, 7, 8))
rdd1.intersection(rdd2)
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a4475b1193be01efc305ef3c39f4b1e8.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a4475b1193be01efc305ef3c39f4b1e8.png" alt="a4475b1193be01efc305ef3c39f4b1e8" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/76a9873eae8de8a9ed5223921da7c245.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/76a9873eae8de8a9ed5223921da7c245.png" alt="76a9873eae8de8a9ed5223921da7c245" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>Intersection 算子是一个集合操作, 用于求得 左侧集合 和 右侧集合 的交集, 换句话说, 就是左侧集合和右侧集合都有的元素, 并生成一个新的 RDD</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>subtract(other, numPartitions)</code></p></th>
<td class="tableblock halign-left valign-top"><p class="tableblock"><strong>(RDD[T], RDD[T]) &#8658; RDD[T]</strong> 差集, 可以设置分区数</p></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>distinct(numPartitions)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">sc.parallelize(Seq(1, 1, 2, 2, 3))
  .distinct()
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a8cd033d9ce502337ba746d05ca94ae1.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a8cd033d9ce502337ba746d05ca94ae1.png" alt="a8cd033d9ce502337ba746d05ca94ae1" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/2bfefe5f5cab497d5aded3b7537a58ba.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/2bfefe5f5cab497d5aded3b7537a58ba.png" alt="2bfefe5f5cab497d5aded3b7537a58ba" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>Distinct 算子用于去重</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>Distinct 是一个需要 Shuffled 的操作</p>
</li>
<li>
<p>本质上 Distinct 就是一个 reductByKey, 把重复的合并为一个</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>reduceByKey((V, V) &#8658; V, numPartition)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
  .reduceByKey( (curr, agg) =&gt; curr + agg )
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a9b444d144d6996c83b33f6a48806a1a.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a9b444d144d6996c83b33f6a48806a1a.png" alt="a9b444d144d6996c83b33f6a48806a1a" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/07678e1b4d6ba1dfaf2f5df89489def4.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/07678e1b4d6ba1dfaf2f5df89489def4.png" alt="07678e1b4d6ba1dfaf2f5df89489def4" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>首先按照 Key 分组生成一个 Tuple, 然后针对每个组执行 <code>reduce</code> 算子</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<p><code>def reduceByKey(func: (V, V) &#8658; V): RDD[(K, V)]</code></p>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>func &#8594; 执行数据处理的函数, 传入两个参数, 一个是当前值, 一个是局部汇总, 这个函数需要有一个输出, 输出就是这个 Key 的汇总结果</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>ReduceByKey 只能作用于 Key-Value 型数据, Key-Value 型数据在当前语境中特指 Tuple2</p>
</li>
<li>
<p>ReduceByKey 是一个需要 Shuffled 的操作</p>
</li>
<li>
<p>和其它的 Shuffled 相比, ReduceByKey是高效的, 因为类似 MapReduce 的, 在 Map 端有一个 Cominer, 这样 I/O 的数据便会减少</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>groupByKey()</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
  .groupByKey()
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/466c1ad2b738c4f0d27f2557ecedaf5b.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/466c1ad2b738c4f0d27f2557ecedaf5b.png" alt="466c1ad2b738c4f0d27f2557ecedaf5b" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/27de81df110abb6709bf1c5ffad184ab.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/27de81df110abb6709bf1c5ffad184ab.png" alt="27de81df110abb6709bf1c5ffad184ab" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>GroupByKey 算子的主要作用是按照 Key 分组, 和 ReduceByKey 有点类似, 但是 GroupByKey 并不求聚合, 只是列举 Key 对应的所有 Value</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>GroupByKey 是一个 Shuffled</p>
</li>
<li>
<p>GroupByKey 和 ReduceByKey 不同, 因为需要列举 Key 对应的所有数据, 所以无法在 Map 端做 Combine, 所以 GroupByKey 的性能并没有 ReduceByKey 好</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>combineByKey()</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd = sc.parallelize(Seq(
  ("zhangsan", 99.0),
  ("zhangsan", 96.0),
  ("lisi", 97.0),
  ("lisi", 98.0),
  ("zhangsan", 97.0))
)

val combineRdd = rdd.combineByKey(
  score =&gt; (score, 1),
  (scoreCount: (Double, Int),newScore) =&gt; (scoreCount._1 + newScore, scoreCount._2 + 1),
  (scoreCount1: (Double, Int), scoreCount2: (Double, Int)) =&gt;
    (scoreCount1._1 + scoreCount2._1, scoreCount1._2 + scoreCount2._2)
)

val meanRdd = combineRdd.map(score =&gt; (score._1, score._2._1 / score._2._2))

meanRdd.collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/741d814a50e4c01686f394df079458bf.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/Snipaste_2019-05-16_16-44-56.png" alt="Snipaste 2019 05 16 16 44 56" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>对数据集按照 Key 进行聚合</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>combineByKey(createCombiner, mergeValue, mergeCombiners, [partitioner], [mapSideCombiner], [serializer])</code></p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>createCombiner</code> 将 Value 进行初步转换</p>
</li>
<li>
<p><code>mergeValue</code> 在每个分区把上一步转换的结果聚合</p>
</li>
<li>
<p><code>mergeCombiners</code> 在所有分区上把每个分区的聚合结果聚合</p>
</li>
<li>
<p><code>partitioner</code> 可选, 分区函数</p>
</li>
<li>
<p><code>mapSideCombiner</code> 可选, 是否在 Map 端 Combine</p>
</li>
<li>
<p><code>serializer</code> 序列化器</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>combineByKey</code> 的要点就是三个函数的意义要理解</p>
</li>
<li>
<p><code>groupByKey</code>, <code>reduceByKey</code> 的底层都是 <code>combineByKey</code></p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>aggregateByKey()</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
val result = rdd.aggregateByKey(0.8)(
  seqOp = (zero, price) =&gt; price * zero,
  combOp = (curr, agg) =&gt; curr + agg
).collect()
println(result)</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ee33b17dbc78705dbbd76d76ab4a9072.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/ee33b17dbc78705dbbd76d76ab4a9072.png" alt="ee33b17dbc78705dbbd76d76ab4a9072" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>聚合所有 Key 相同的 Value, 换句话说, 按照 Key 聚合 Value</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>rdd.aggregateByKey(zeroValue)(seqOp, combOp)</code></p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>zeroValue</code> 初始值</p>
</li>
<li>
<p><code>seqOp</code> 转换每一个值的函数</p>
</li>
<li>
<p><code>comboOp</code> 将转换过的值聚合的函数</p>
</li>
</ul>
</div>
</dd>
</dl>
</div>
<div class="paragraph">
<p>注意点<strong>
* 为什么需要两个函数?
</strong> aggregateByKey 运行将一个`RDD[(K, V)]<code>聚合为`RDD[(K, U)]</code>, 如果要做到这件事的话, 就需要先对数据做一次转换, 将每条数据从`V`转为`U`, `seqOp`就是干这件事的
** 当`seqOp`的事情结束以后, `comboOp`把其结果聚合</p>
</div>
<div class="ulist">
<ul>
<li>
<p>和 reduceByKey 的区别::</p>
<div class="ulist">
<ul>
<li>
<p>aggregateByKey 最终聚合结果的类型和传入的初始值类型保持一致</p>
</li>
<li>
<p>reduceByKey 在集合中选取第一个值作为初始值, 并且聚合过的数据类型不能改变</p>
</li>
</ul>
</div>
</li>
</ul>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>foldByKey(zeroValue)((V, V) &#8658; V)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">sc.parallelize(Seq(("a", 1), ("a", 1), ("b", 1)))
  .foldByKey(zeroValue = 10)( (curr, agg) =&gt; curr + agg )
  .collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/c00063a109a0f9e0b1c2b385c5e1cc47.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/c00063a109a0f9e0b1c2b385c5e1cc47.png" alt="c00063a109a0f9e0b1c2b385c5e1cc47" width="800"></a>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a406ff8395bb092e719007661b34d385.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a406ff8395bb092e719007661b34d385.png" alt="a406ff8395bb092e719007661b34d385" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>和 ReduceByKey 是一样的, 都是按照 Key 做分组去求聚合, 但是 FoldByKey 的不同点在于可以指定初始值</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<p><code>foldByKey(zeroValue)(func)</code></p>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>zeroValue</code> 初始值</p>
</li>
<li>
<p><code>func</code> seqOp 和 combOp 相同, 都是这个参数</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>FoldByKey 是 AggregateByKey 的简化版本,  seqOp 和 combOp 是同一个函数</p>
</li>
<li>
<p>FoldByKey 指定的初始值作用于每一个 Value</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>join(other, numPartitions)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("b", 1)))
val rdd2 = sc.parallelize(Seq(("a", 10), ("a", 11), ("a", 12)))

rdd1.join(rdd2).collect()</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/bb3eda1410d3b0f6e1bff6d5e6a45879.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/bb3eda1410d3b0f6e1bff6d5e6a45879.png" alt="bb3eda1410d3b0f6e1bff6d5e6a45879" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>将两个 RDD 按照相同的 Key 进行连接</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<p><code>join(other, [partitioner or numPartitions])</code></p>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>other</code> 其它 RDD</p>
</li>
<li>
<p><code>partitioner or numPartitions</code> 可选, 可以通过传递分区函数或者分区数量来改变分区</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>Join 有点类似于 SQL 中的内连接, 只会再结果中包含能够连接到的 Key</p>
</li>
<li>
<p>Join 的结果是一个笛卡尔积形式, 例如`"a", 1), ("a", 2<code>和</code>"a", 10), ("a", 11<code>的 Join 结果集是 `"a", 1, 10), ("a", 1, 11), ("a", 2, 10), ("a", 2, 11</code></p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>cogroup(other, numPartitions)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd1 = sc.parallelize(Seq(("a", 1), ("a", 2), ("a", 5), ("b", 2), ("b", 6), ("c", 3), ("d", 2)))
val rdd2 = sc.parallelize(Seq(("a", 10), ("b", 1), ("d", 3)))
val rdd3 = sc.parallelize(Seq(("b", 10), ("a", 1)))

val result1 = rdd1.cogroup(rdd2).collect()
val result2 = rdd1.cogroup(rdd2, rdd3).collect()

/*
执行结果:
Array(
  (d,(CompactBuffer(2),CompactBuffer(3))),
  (a,(CompactBuffer(1, 2, 5),CompactBuffer(10))),
  (b,(CompactBuffer(2, 6),CompactBuffer(1))),
  (c,(CompactBuffer(3),CompactBuffer()))
)
 */
println(result1)

/*
执行结果:
Array(
  (d,(CompactBuffer(2),CompactBuffer(3),CompactBuffer())),
  (a,(CompactBuffer(1, 2, 5),CompactBuffer(10),CompactBuffer(1))),
  (b,(CompactBuffer(2, 6),CompactBuffer(1),Co...
 */
println(result2)</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<a class="image" href="https://doc-1256053707.cos.ap-beijing.myqcloud.com/42262ffe7f3ff35013fbe534d78e3518.png"><img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/42262ffe7f3ff35013fbe534d78e3518.png" alt="42262ffe7f3ff35013fbe534d78e3518" width="800"></a>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>多个 RDD 协同分组, 将多个 RDD 中 Key 相同的 Value 分组</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>cogroup(rdd1, rdd2, rdd3, [partitioner or numPartitions])</code></p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>rdd&#8230;&#8203;</code> 最多可以传三个 RDD 进去, 加上调用者, 可以为四个 RDD 协同分组</p>
</li>
<li>
<p><code>partitioner or numPartitions</code> 可选, 可以通过传递分区函数或者分区数来改变分区</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>对 RDD1, RDD2, RDD3 进行 cogroup, 结果中就一定会有三个 List, 如果没有 Value 则是空 List, 这一点类似于 SQL 的全连接, 返回所有结果, 即使没有关联上</p>
</li>
<li>
<p>CoGroup 是一个需要 Shuffled 的操作</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>cartesian(other)</code></p></th>
<td class="tableblock halign-left valign-top"><p class="tableblock"><strong>(RDD[T], RDD[U]) &#8658; RDD[(T, U)]</strong> 生成两个 RDD 的笛卡尔积</p></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>sortBy(ascending, numPartitions)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd1 = sc.parallelize(Seq(("a", 3), ("b", 2), ("c", 1)))
val sortByResult = rdd1.sortBy( item =&gt; item._2 ).collect()
val sortByKeyResult = rdd1.sortByKey().collect()

println(sortByResult)
println(sortByKeyResult)</code></pre>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>排序相关相关的算子有两个, 一个是`sortBy`, 另外一个是`sortByKey`</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<p><code>sortBy(func, ascending, numPartitions)</code></p>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>`func`通过这个函数返回要排序的字段</p>
</li>
<li>
<p>`ascending`是否升序</p>
</li>
<li>
<p>`numPartitions`分区数</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>普通的 RDD 没有`sortByKey`, 只有 Key-Value 的 RDD 才有</p>
</li>
<li>
<p>`sortBy`可以指定按照哪个字段来排序, `sortByKey`直接按照 Key 来排序</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>partitionBy(partitioner)</code></p></th>
<td class="tableblock halign-left valign-top"><p class="tableblock">使用用传入的 partitioner 重新分区, 如果和当前分区函数相同, 则忽略操作</p></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>coalesce(numPartitions)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>减少分区数</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd = sc.parallelize(Seq(("a", 3), ("b", 2), ("c", 1)))
val oldNum = rdd.partitions.length

val coalesceRdd = rdd.coalesce(4, shuffle = true)
val coalesceNum = coalesceRdd.partitions.length

val repartitionRdd = rdd.repartition(4)
val repartitionNum = repartitionRdd.partitions.length

print(oldNum, coalesceNum, repartitionNum)</code></pre>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>一般涉及到分区操作的算子常见的有两个, <code>repartitioin</code> 和 <code>coalesce</code>, 两个算子都可以调大或者调小分区数量</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>repartitioin(numPartitions)</code></p>
</li>
<li>
<p><code>coalesce(numPartitions, shuffle)</code></p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">参数</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>numPartitions</code> 新的分区数</p>
</li>
<li>
<p><code>shuffle</code> 是否 shuffle, 如果新的分区数量比原分区数大, 必须 Shuffled, 否则重分区无效</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>repartition</code> 和 <code>coalesce</code> 的不同就在于 <code>coalesce</code> 可以控制是否 Shuffle</p>
</li>
<li>
<p><code>repartition</code> 是一个 Shuffled 操作</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>repartition(numPartitions)</code></p></th>
<td class="tableblock halign-left valign-top"><p class="tableblock">重新分区</p></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>repartitionAndSortWithinPartitions</code></p></th>
<td class="tableblock halign-left valign-top"><p class="tableblock">重新分区的同时升序排序, 在partitioner中排序, 比先重分区再排序要效率高, 建议使用在需要分区后再排序的场景使用</p></td>
</tr>
</tbody>
</table>
</div>
<div class="sect2">
<h3 id="_2_2_action_算子">2.2. Action 算子</h3>
<table class="tableblock frame-all grid-all stretch">
<colgroup>
<col style="width: 25%;">
<col>
</colgroup>
<thead>
<tr>
<th class="tableblock halign-left valign-top">Action function</th>
<th class="tableblock halign-left valign-top">解释</th>
</tr>
</thead>
<tbody>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>reduce( (T, T) &#8658; U )</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
val result = rdd.reduce((curr, agg) =&gt; ("总价", curr._2 + agg._2))
println(result)</code></pre>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>对整个结果集规约, 最终生成一条数据, 是整个数据集的汇总</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">调用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>reduce( (currValue[T], agg[T]) &#8658; T )</code></p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>reduce 和 reduceByKey 是完全不同的, reduce 是一个 action, 并不是 Shuffled 操作</p>
</li>
<li>
<p>本质上 reduce 就是现在每个 partition 上求值, 最终把每个 partition 的结果再汇总</p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>collect()</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>以数组的形式返回数据集中所有元素</p>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>count()</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>返回元素个数</p>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>first()</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>返回第一个元素</p>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>take( N )</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>返回前 N 个元素</p>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>takeSample(withReplacement, fract)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>类似于 sample, 区别在这是一个Action, 直接返回结果</p>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>fold(zeroValue)( (T, T) &#8658; U )</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>指定初始值和计算函数, 折叠聚合整个数据集</p>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>saveAsTextFile(path)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>将结果存入 path 对应的文件中</p>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>saveAsSequenceFile(path)</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>将结果存入 path 对应的 Sequence 文件中</p>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>countByKey()</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
val result = rdd.countByKey()
println(result)</code></pre>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">作用</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>求得整个数据集中 Key 以及对应 Key 出现的次数</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">注意点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>返回结果为 <code>Map(key &#8594; count)</code></p>
</li>
<li>
<p><strong>常在解决数据倾斜问题时使用, 查看倾斜的 Key</strong></p>
</li>
</ul>
</div>
</dd>
</dl>
</div></div></td>
</tr>
<tr>
<th class="tableblock halign-left valign-top"><p class="tableblock"><code>foreach( T &#8658; &#8230;&#8203; )</code></p></th>
<td class="tableblock halign-left valign-top"><div class="content"><div class="paragraph">
<p>遍历每一个元素</p>
</div></div></td>
</tr>
</tbody>
</table>
<div class="dlist">
<dl>
<dt class="hdlist1">应用</dt>
<dd>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">```scala
val rdd = sc.parallelize(Seq(("手机", 10.0), ("手机", 15.0), ("电脑", 20.0)))
// 结果: Array((手机,10.0), (手机,15.0), (电脑,20.0))
println(rdd.collect())
// 结果: Array((手机,10.0), (手机,15.0))
println(rdd.take(2))
// 结果: (手机,10.0)
println(rdd.first())
```</code></pre>
</div>
</div>
</dd>
</dl>
</div>
<div class="exampleblock">
<div class="title">总结</div>
<div class="content">
<div class="dlist">
<dl>
<dt class="hdlist1">RDD 的算子大部分都会生成一些专用的 RDD</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p><code>map</code>, <code>flatMap</code>, <code>filter</code> 等算子会生成 <code>MapPartitionsRDD</code></p>
</li>
<li>
<p><code>coalesce</code>, <code>repartition</code> 等算子会生成 <code>CoalescedRDD</code></p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">常见的 RDD 有两种类型</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>转换型的 RDD, Transformation</p>
</li>
<li>
<p>动作型的 RDD, Action</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">常见的 Transformation 类型的 RDD</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>map</p>
</li>
<li>
<p>flatMap</p>
</li>
<li>
<p>filter</p>
</li>
<li>
<p>groupBy</p>
</li>
<li>
<p>reduceByKey</p>
</li>
</ul>
</div>
</dd>
<dt class="hdlist1">常见的 Action 类型的 RDD</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>collect</p>
</li>
<li>
<p>countByKey</p>
</li>
<li>
<p>reduce</p>
</li>
</ul>
</div>
</dd>
</dl>
</div>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_2_3_rdd_对不同类型数据的支持">2.3. RDD 对不同类型数据的支持</h3>
<div class="exampleblock">
<div class="title">目标</div>
<div class="content">
<div class="olist arabic">
<ol class="arabic">
<li>
<p>理解 RDD 对 Key-Value 类型的数据是有专门支持的</p>
</li>
<li>
<p>理解 RDD 对数字类型也有专门的支持</p>
</li>
</ol>
</div>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">一般情况下 RDD 要处理的数据有三类</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="ulist">
<ul>
<li>
<p>字符串</p>
</li>
<li>
<p>键值对</p>
</li>
<li>
<p>数字型</p>
</li>
</ul>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">RDD 的算子设计对这三类不同的数据分别都有支持</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="ulist">
<ul>
<li>
<p>对于以字符串为代表的基本数据类型是比较基础的一些的操作, 诸如 map, flatMap, filter 等基础的算子</p>
</li>
<li>
<p>对于键值对类型的数据, 有额外的支持, 诸如 reduceByKey, groupByKey 等 byKey 的算子</p>
</li>
<li>
<p>同样对于数字型的数据也有额外的支持, 诸如 max, min 等</p>
</li>
</ul>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">RDD 对键值对数据的额外支持</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="paragraph">
<p>键值型数据本质上就是一个二元元组, 键值对类型的 RDD 表示为 <code>RDD[(K, V)]</code></p>
</div>
<div class="paragraph">
<p>RDD 对键值对的额外支持是通过隐式支持来完成的, 一个 <code>RDD[(K, V)]</code>, 可以被隐式转换为一个 <code>PairRDDFunctions</code> 对象, 从而调用其中的方法.</p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/3b365c28403495cb8d07a2ee5d0a6376.png" alt="3b365c28403495cb8d07a2ee5d0a6376" width="800">
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">既然对键值对的支持是通过 <code>PairRDDFunctions</code> 提供的, 那么从 <code>PairRDDFunctions</code> 中就可以看到这些支持有什么</dt>
</dl>
</div>
<table class="tableblock frame-all grid-all stretch">
<colgroup>
<col style="width: 50%;">
<col style="width: 50%;">
</colgroup>
<thead>
<tr>
<th class="tableblock halign-left valign-top">类别</th>
<th class="tableblock halign-left valign-top">算子</th>
</tr>
</thead>
<tbody>
<tr>
<td class="tableblock halign-left valign-top" rowspan="3"><p class="tableblock">聚合操作</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>reduceByKey</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>foldByKey</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>combineByKey</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top" rowspan="2"><p class="tableblock">分组操作</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>cogroup</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>groupByKey</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top" rowspan="3"><p class="tableblock">连接操作</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>join</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>leftOuterJoin</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>rightOuterJoin</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top" rowspan="2"><p class="tableblock">排序操作</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>sortBy</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>sortByKey</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top" rowspan="3"><p class="tableblock">Action</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>countByKey</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>take</code></p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>collect</code></p></td>
</tr>
</tbody>
</table>
</div>
</div>
</dd>
<dt class="hdlist1">RDD 对数字型数据的额外支持</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="paragraph">
<p>对于数字型数据的额外支持基本上都是 Action 操作, 而不是转换操作</p>
</div>
<table class="tableblock frame-all grid-all stretch">
<colgroup>
<col style="width: 50%;">
<col style="width: 50%;">
</colgroup>
<thead>
<tr>
<th class="tableblock halign-left valign-top">算子</th>
<th class="tableblock halign-left valign-top">含义</th>
</tr>
</thead>
<tbody>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>count</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">个数</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>mean</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">均值</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>sum</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">求和</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>max</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">最大值</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>min</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">最小值</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>variance</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">方差</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>sampleVariance</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">从采样中计算方差</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>stdev</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">标准差</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>sampleStdev</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">采样的标准差</p></td>
</tr>
</tbody>
</table>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val rdd = sc.parallelize(Seq(1, 2, 3))
// 结果: 3
println(rdd.max())</code></pre>
</div>
</div>
</div>
</div>
</dd>
</dl>
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="_3_rdd_的_shuffle_和分区">3. RDD 的 Shuffle 和分区</h2>
<div class="sectionbody">
<div class="exampleblock">
<div class="title">目标</div>
<div class="content">
<div class="olist arabic">
<ol class="arabic">
<li>
<p>RDD 的分区操作</p>
</li>
<li>
<p>Shuffle 的原理</p>
</li>
</ol>
</div>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">分区的作用</dt>
<dd>
<p>RDD 使用分区来分布式并行处理数据, 并且要做到尽量少的在不同的 Executor 之间使用网络交换数据, 所以当使用 RDD 读取数据的时候, 会尽量的在物理上靠近数据源, 比如说在读取 Cassandra 或者 HDFS 中数据的时候, 会尽量的保持 RDD 的分区和数据源的分区数, 分区模式等一一对应</p>
</dd>
<dt class="hdlist1">分区和 Shuffle 的关系</dt>
<dd>
<p>分区的主要作用是用来实现并行计算, 本质上和 Shuffle 没什么关系, 但是往往在进行数据处理的时候, 例如`reduceByKey`, `groupByKey`等聚合操作, 需要把 Key 相同的 Value 拉取到一起进行计算, 这个时候因为这些 Key 相同的 Value 可能会坐落于不同的分区, 于是理解分区才能理解 Shuffle 的根本原理</p>
</dd>
<dt class="hdlist1">Spark 中的 Shuffle 操作的特点</dt>
<dd>
<div class="ulist">
<ul>
<li>
<p>只有 <code>Key-Value</code> 型的 RDD 才会有 Shuffle 操作, 例如 <code>RDD[(K, V)]</code>, 但是有一个特例, 就是 <code>repartition</code> 算子可以对任何数据类型 Shuffle</p>
</li>
<li>
<p>早期版本 Spark 的 Shuffle 算法是 <code>Hash base shuffle</code>, 后来改为 <code>Sort base shuffle</code>, 更适合大吞吐量的场景</p>
</li>
</ul>
</div>
</dd>
</dl>
</div>
<div class="sect2">
<h3 id="_3_1_rdd_的分区操作">3.1. RDD 的分区操作</h3>
<div class="dlist">
<dl>
<dt class="hdlist1">查看分区数</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-java hljs" data-lang="java">scala&gt; sc.parallelize(1 to 100).count
res0: Long = 100</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/873af6194db362a1ab5432372aa8bd21.png" alt="873af6194db362a1ab5432372aa8bd21" width="800">
</div>
</div>
<div class="paragraph">
<p>之所以会有 8 个 Tasks, 是因为在启动的时候指定的命令是 <code>spark-shell --master local[8]</code>, 这样会生成 1 个 Executors, 这个 Executors 有 8 个 Cores,
所以默认会有 8 个 Tasks, 每个 Cores 对应一个分区, 每个分区对应一个 Tasks, 可以通过 <code>rdd.partitions.size</code> 来查看分区数量</p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/a41901e5af14f37c88b3f1ea9b97fbfb.png" alt="a41901e5af14f37c88b3f1ea9b97fbfb" width="800">
</div>
</div>
<div class="paragraph">
<p>同时也可以通过 spark-shell 的 WebUI 来查看 Executors 的情况</p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/24b2646308923d7549a7758f7550e0a8.png" alt="24b2646308923d7549a7758f7550e0a8" width="800">
</div>
</div>
</div>
</div>
</dd>
</dl>
</div>
<div class="paragraph">
<p>默认的分区数量是和 Cores 的数量有关的, 也可以通过如下三种方式修改或者重新指定分区数量</p>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">创建 RDD 时指定分区数</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-java hljs" data-lang="java">scala&gt; val rdd1 = sc.parallelize(1 to 100, 6)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at &lt;console&gt;:24

scala&gt; rdd1.partitions.size
res1: Int = 6

scala&gt; val rdd2 = sc.textFile("hdfs:///dataset/wordcount.txt", 6)
rdd2: org.apache.spark.rdd.RDD[String] = hdfs:///dataset/wordcount.txt MapPartitionsRDD[3] at textFile at &lt;console&gt;:24

scala&gt; rdd2.partitions.size
res2: Int = 7</code></pre>
</div>
</div>
<div class="paragraph">
<p>rdd1 是通过本地集合创建的, 创建的时候通过第二个参数指定了分区数量. rdd2 是通过读取 HDFS 中文件创建的, 同样通过第二个参数指定了分区数,
因为是从 HDFS 中读取文件, 所以最终的分区数是由 Hadoop 的 InputFormat 来指定的, 所以比指定的分区数大了一个.</p>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">通过`coalesce` 算子指定</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-java hljs" data-lang="java">coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = null): RDD[T]</code></pre>
</div>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">numPartitions</dt>
<dd>
<p>新生成的 RDD 的分区数</p>
</dd>
<dt class="hdlist1">shuffle</dt>
<dd>
<p>是否 Shuffle</p>
</dd>
</dl>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code>scala&gt; val source = sc.parallelize(1 to 100, 6)
source: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at &lt;console&gt;:24

scala&gt; source.partitions.size
res0: Int = 6

scala&gt; val noShuffleRdd = source.coalesce(numPartitions=8, shuffle=false)
noShuffleRdd: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[1] at coalesce at &lt;console&gt;:26

scala&gt; noShuffleRdd.toDebugString <i class="conum" data-value="1"></i><b>(1)</b>
res1: String =
(6) CoalescedRDD[1] at coalesce at &lt;console&gt;:26 []
 |  ParallelCollectionRDD[0] at parallelize at &lt;console&gt;:24 []

 scala&gt; val noShuffleRdd = source.coalesce(numPartitions=8, shuffle=false)
 noShuffleRdd: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[1] at coalesce at &lt;console&gt;:26

scala&gt; shuffleRdd.toDebugString <i class="conum" data-value="2"></i><b>(2)</b>
res3: String =
(8) MapPartitionsRDD[5] at coalesce at &lt;console&gt;:26 []
 |  CoalescedRDD[4] at coalesce at &lt;console&gt;:26 []
 |  ShuffledRDD[3] at coalesce at &lt;console&gt;:26 []
 +-(6) MapPartitionsRDD[2] at coalesce at &lt;console&gt;:26 []
    |  ParallelCollectionRDD[0] at parallelize at &lt;console&gt;:24 []

scala&gt; noShuffleRdd.partitions.size     <i class="conum" data-value="3"></i><b>(3)</b>
res4: Int = 6

scala&gt; shuffleRdd.partitions.size
res5: Int = 8</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>如果 <code>shuffle</code> 参数指定为 <code>false</code>, 运行计划中确实没有 <code>ShuffledRDD</code>, 没有 <code>shuffled</code> 这个过程</td>
</tr>
<tr>
<td><i class="conum" data-value="2"></i><b>2</b></td>
<td>如果 <code>shuffle</code> 参数指定为 <code>true</code>, 运行计划中有一个 <code>ShuffledRDD</code>, 有一个明确的显式的 <code>shuffled</code> 过程</td>
</tr>
<tr>
<td><i class="conum" data-value="3"></i><b>3</b></td>
<td>如果 <code>shuffle</code> 参数指定为 <code>false</code> 却增加了分区数, 分区数并不会发生改变, 这是因为增加分区是一个宽依赖, 没有 <code>shuffled</code> 过程无法做到, 后续会详细解释宽依赖的概念</td>
</tr>
</table>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">通过 <code>repartition</code> 算子指定</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-java hljs" data-lang="java">repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]</code></pre>
</div>
</div>
<div class="paragraph">
<p><code>repartition</code> 算子本质上就是 <code>coalesce(numPartitions, shuffle = true)</code></p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/45d7a2b6e9e2727504e9cf28adbe6c49.png" alt="45d7a2b6e9e2727504e9cf28adbe6c49" width="800">
</div>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code>scala&gt; val source = sc.parallelize(1 to 100, 6)
source: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[7] at parallelize at &lt;console&gt;:24

scala&gt; source.partitions.size
res7: Int = 6

scala&gt; source.repartition(100).partitions.size <i class="conum" data-value="1"></i><b>(1)</b>
res8: Int = 100

scala&gt; source.repartition(1).partitions.size <i class="conum" data-value="2"></i><b>(2)</b>
res9: Int = 1</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>增加分区有效</td>
</tr>
<tr>
<td><i class="conum" data-value="2"></i><b>2</b></td>
<td>减少分区有效</td>
</tr>
</table>
</div>
<div class="paragraph">
<p><code>repartition</code> 算子无论是增加还是减少分区都是有效的, 因为本质上 <code>repartition</code> 会通过 <code>shuffle</code> 操作把数据分发给新的 RDD 的不同的分区, 只有 <code>shuffle</code> 操作才可能做到增大分区数,
默认情况下, 分区函数是 <code>RoundRobin</code>, 如果希望改变分区函数, 也就是数据分布的方式, 可以通过自定义分区函数来实现</p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/b1181258789202436ca6d2d92e604d59.png" alt="b1181258789202436ca6d2d92e604d59" width="800">
</div>
</div>
</div>
</div>
</dd>
</dl>
</div>
</div>
<div class="sect2">
<h3 id="_3_2_rdd_的_shuffle_是什么">3.2. RDD 的 Shuffle 是什么</h3>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-java hljs" data-lang="java">val sourceRdd = sc.textFile("hdfs://node01:9020/dataset/wordcount.txt")
val flattenCountRdd = sourceRdd.flatMap(_.split(" ")).map((_, 1))
val aggCountRdd = flattenCountRdd.reduceByKey(_ + _)
val result = aggCountRdd.collect</code></pre>
</div>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/23377ac4a368fc94b6f8f3117af67154.png" alt="23377ac4a368fc94b6f8f3117af67154" width="800">
</div>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/10b536c17409ec37fa1f1b308b2b521e.png" alt="10b536c17409ec37fa1f1b308b2b521e" width="800">
</div>
</div>
<div class="paragraph">
<p><code>reduceByKey</code> 这个算子本质上就是先按照 Key 分组, 后对每一组数据进行 <code>reduce</code>, 所面临的挑战就是 Key 相同的所有数据可能分布在不同的 Partition 分区中, 甚至可能在不同的节点中, 但是它们必须被共同计算.</p>
</div>
<div class="paragraph">
<p>为了让来自相同 Key 的所有数据都在 <code>reduceByKey</code> 的同一个 <code>reduce</code> 中处理, 需要执行一个 <code>all-to-all</code> 的操作, 需要在不同的节点(不同的分区)之间拷贝数据, 必须跨分区聚集相同 Key 的所有数据, 这个过程叫做 <code>Shuffle</code>.</p>
</div>
</div>
<div class="sect2">
<h3 id="_3_3_rdd_的_shuffle_原理">3.3. RDD 的 Shuffle 原理</h3>
<div class="paragraph">
<p>Spark 的 Shuffle 发展大致有两个阶段: <code>Hash base shuffle</code> 和 <code>Sort base shuffle</code></p>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">Hash base shuffle</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/2daf43cc1750fffab62ae5e16fab54c2.png" alt="2daf43cc1750fffab62ae5e16fab54c2">
</div>
</div>
<div class="paragraph">
<p>大致的原理是分桶, 假设 Reducer 的个数为 R, 那么每个 Mapper 有 R 个桶, 按照 Key 的 Hash 将数据映射到不同的桶中, Reduce 找到每一个 Mapper 中对应自己的桶拉取数据.</p>
</div>
<div class="paragraph">
<p>假设 Mapper 的个数为 M, 整个集群的文件数量是 <code>M * R</code>, 如果有 1,000 个 Mapper 和 Reducer, 则会生成 1,000,000 个文件, 这个量非常大了.</p>
</div>
<div class="paragraph">
<p>过多的文件会导致文件系统打开过多的文件描述符, 占用系统资源. 所以这种方式并不适合大规模数据的处理, 只适合中等规模和小规模的数据处理, 在 Spark 1.2 版本中废弃了这种方式.</p>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">Sort base shuffle</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/94f038994f8553dd32370ae78878d038.png" alt="94f038994f8553dd32370ae78878d038">
</div>
</div>
<div class="paragraph">
<p>对于 Sort base shuffle 来说, 每个 Map 侧的分区只有一个输出文件, Reduce 侧的 Task 来拉取, 大致流程如下</p>
</div>
<div class="olist arabic">
<ol class="arabic">
<li>
<p>Map 侧将数据全部放入一个叫做 AppendOnlyMap 的组件中, 同时可以在这个特殊的数据结构中做聚合操作</p>
</li>
<li>
<p>然后通过一个类似于 MergeSort 的排序算法 TimSort 对 AppendOnlyMap 底层的 Array 排序</p>
<div class="ulist">
<ul>
<li>
<p>先按照 Partition ID 排序, 后按照 Key 的 HashCode 排序</p>
</li>
</ul>
</div>
</li>
<li>
<p>最终每个 Map Task 生成一个 输出文件, Reduce Task 来拉取自己对应的数据</p>
</li>
</ol>
</div>
<div class="paragraph">
<p>从上面可以得到结论, Sort base shuffle 确实可以大幅度减少所产生的中间文件, 从而能够更好的应对大吞吐量的场景, 在 Spark 1.2 以后, 已经默认采用这种方式.</p>
</div>
<div class="paragraph">
<p>但是需要大家知道的是, Spark 的 Shuffle 算法并不只是这一种, 即使是在最新版本, 也有三种 Shuffle 算法, 这三种算法对每个 Map 都只产生一个临时文件, 但是产生文件的方式不同, 一种是类似 Hash 的方式, 一种是刚才所说的 Sort, 一种是对 Sort 的一种优化(使用 Unsafe API 直接申请堆外内存)</p>
</div>
</div>
</div>
</dd>
</dl>
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="_4_缓存">4. 缓存</h2>
<div class="sectionbody">
<div class="exampleblock">
<div class="title">概要</div>
<div class="content">
<div class="olist arabic">
<ol class="arabic">
<li>
<p>缓存的意义</p>
</li>
<li>
<p>缓存相关的 API</p>
</li>
<li>
<p>缓存级别以及最佳实践</p>
</li>
</ol>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_4_1_缓存的意义">4.1. 缓存的意义</h3>
<div class="dlist">
<dl>
<dt class="hdlist1">使用缓存的原因 - 多次使用 RDD</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="paragraph">
<p>需求: 在日志文件中找到访问次数最少的 IP 和访问次数最多的 IP</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
  .map(item =&gt; (item.split(" ")(0), 1))
  .filter(item =&gt; StringUtils.isNotBlank(item._1))
  .reduceByKey((curr, agg) =&gt; curr + agg) <i class="conum" data-value="1"></i><b>(1)</b>

val resultLess = interimRDD.sortBy(item =&gt; item._2, ascending = true).first()
val resultMore = interimRDD.sortBy(item =&gt; item._2, ascending = false).first()

println(s"出现次数最少的 IP : $resultLess, 出现次数最多的 IP : $resultMore")

sc.stop()</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>这是一个 Shuffle 操作, Shuffle 操作会在集群内进行数据拷贝</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>在上述代码中, 多次使用到了 <code>interimRDD</code>, 导致文件读取两次, 计算两次, 有没有什么办法增进上述代码的性能?</p>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">使用缓存的原因 - 容错</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/20190511163654.png" alt="20190511163654">
</div>
</div>
<div class="paragraph">
<p>当在计算 RDD3 的时候如果出错了, 会怎么进行容错?</p>
</div>
<div class="paragraph">
<p>会再次计算 RDD1 和 RDD2 的整个链条, 假设 RDD1 和 RDD2 是通过比较昂贵的操作得来的, 有没有什么办法减少这种开销?</p>
</div>
</div>
</div>
</dd>
</dl>
</div>
<div class="paragraph">
<p>上述两个问题的解决方案其实都是 <code>缓存</code>, 除此之外, 使用缓存的理由还有很多, 但是总结一句, 就是缓存能够帮助开发者在进行一些昂贵操作后, 将其结果保存下来, 以便下次使用无需再次执行, 缓存能够显著的提升性能.</p>
</div>
<div class="paragraph">
<p>所以, 缓存适合在一个 RDD 需要重复多次利用, 并且还不是特别大的情况下使用, 例如迭代计算等场景.</p>
</div>
</div>
<div class="sect2">
<h3 id="_4_2_缓存相关的_api">4.2. 缓存相关的 API</h3>
<div class="dlist">
<dl>
<dt class="hdlist1">可以使用 <code>cache</code> 方法进行缓存</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
  .map(item =&gt; (item.split(" ")(0), 1))
  .filter(item =&gt; StringUtils.isNotBlank(item._1))
  .reduceByKey((curr, agg) =&gt; curr + agg)
  .cache() <i class="conum" data-value="1"></i><b>(1)</b>

val resultLess = interimRDD.sortBy(item =&gt; item._2, ascending = true).first()
val resultMore = interimRDD.sortBy(item =&gt; item._2, ascending = false).first()

println(s"出现次数最少的 IP : $resultLess, 出现次数最多的 IP : $resultMore")

sc.stop()</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>缓存</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>方法签名如下</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">cache(): this.type = persist()</code></pre>
</div>
</div>
<div class="paragraph">
<p>cache 方法其实是 <code>persist</code> 方法的一个别名</p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/20190511164152.png" alt="20190511164152">
</div>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">也可以使用 persist 方法进行缓存</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="listingblock">
<div class="content">
<pre>val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
  .map(item =&gt; (item.split(" ")(0), 1))
  .filter(item =&gt; StringUtils.isNotBlank(item._1))
  .reduceByKey((curr, agg) =&gt; curr + agg)
  .persist(StorageLevel.MEMORY_ONLY) <i class="conum" data-value="1"></i><b>(1)</b>

val resultLess = interimRDD.sortBy(item =&gt; item._2, ascending = true).first()
val resultMore = interimRDD.sortBy(item =&gt; item._2, ascending = false).first()

println(s"出现次数最少的 IP : $resultLess, 出现次数最多的 IP : $resultMore")

sc.stop()</pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>缓存</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>方法签名如下</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">persist(): this.type
persist(newLevel: StorageLevel): this.type</code></pre>
</div>
</div>
<div class="paragraph">
<p><code>persist</code> 方法其实有两种形式, <code>persist()</code> 是 <code>persist(newLevel: StorageLevel)</code> 的一个别名, <code>persist(newLevel: StorageLevel)</code> 能够指定缓存的级别</p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/20190511164532.png" alt="20190511164532">
</div>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">缓存其实是一种空间换时间的做法, 会占用额外的存储资源, 如何清理?</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
  .map(item =&gt; (item.split(" ")(0), 1))
  .filter(item =&gt; StringUtils.isNotBlank(item._1))
  .reduceByKey((curr, agg) =&gt; curr + agg)
  .persist()

interimRDD.unpersist() <i class="conum" data-value="1"></i><b>(1)</b>

val resultLess = interimRDD.sortBy(item =&gt; item._2, ascending = true).first()
val resultMore = interimRDD.sortBy(item =&gt; item._2, ascending = false).first()

println(s"出现次数最少的 IP : $resultLess, 出现次数最多的 IP : $resultMore")

sc.stop()</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>清理缓存</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>根据缓存级别的不同, 缓存存储的位置也不同, 但是使用 <code>unpersist</code> 可以指定删除 RDD 对应的缓存信息, 并指定缓存级别为 <code>NONE</code></p>
</div>
</div>
</div>
</dd>
</dl>
</div>
</div>
<div class="sect2">
<h3 id="_4_3_缓存级别">4.3. 缓存级别</h3>
<div class="paragraph">
<p>其实如何缓存是一个技术活, 有很多细节需要思考, 如下</p>
</div>
<div class="ulist">
<ul>
<li>
<p>是否使用磁盘缓存?</p>
</li>
<li>
<p>是否使用内存缓存?</p>
</li>
<li>
<p>是否使用堆外内存?</p>
</li>
<li>
<p>缓存前是否先序列化?</p>
</li>
<li>
<p>是否需要有副本?</p>
</li>
</ul>
</div>
<div class="paragraph">
<p>如果要回答这些信息的话, 可以先查看一下 RDD 的缓存级别对象</p>
</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
  .map(item =&gt; (item.split(" ")(0), 1))
  .filter(item =&gt; StringUtils.isNotBlank(item._1))
  .reduceByKey((curr, agg) =&gt; curr + agg)
  .persist()

println(interimRDD.getStorageLevel)

sc.stop()</code></pre>
</div>
</div>
<div class="paragraph">
<p>打印出来的对象是 <code>StorageLevel</code>, 其中有如下几个构造参数</p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/20190511170124.png" alt="20190511170124" width="600">
</div>
</div>
<div class="paragraph">
<p>根据这几个参数的不同, <code>StorageLevel</code> 有如下几个枚举对象</p>
</div>
<div class="imageblock">
<div class="content">
<img src="https://doc-1256053707.cos.ap-beijing.myqcloud.com/20190511170338.png" alt="20190511170338" width="600">
</div>
</div>
<table class="tableblock frame-all grid-all stretch">
<colgroup>
<col style="width: 16.6666%;">
<col style="width: 16.6666%;">
<col style="width: 16.6666%;">
<col style="width: 16.6666%;">
<col style="width: 16.6666%;">
<col style="width: 16.667%;">
</colgroup>
<thead>
<tr>
<th class="tableblock halign-left valign-top">缓存级别</th>
<th class="tableblock halign-left valign-top"><code>userDisk</code> 是否使用磁盘</th>
<th class="tableblock halign-left valign-top"><code>useMemory</code> 是否使用内存</th>
<th class="tableblock halign-left valign-top"><code>useOffHeap</code> 是否使用堆外内存</th>
<th class="tableblock halign-left valign-top"><code>deserialized</code> 是否以反序列化形式存储</th>
<th class="tableblock halign-left valign-top"><code>replication</code> 副本数</th>
</tr>
</thead>
<tbody>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>NONE</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">1</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>DISK_ONLY</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">1</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>DISK_ONLY_2</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">2</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>MEMORY_ONLY</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">1</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>MEMORY_ONLY_2</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">2</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>MEMORY_ONLY_SER</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">1</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>MEMORY_ONLY_SER_2</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">2</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>MEMORY_AND_DISK</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">1</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>MEMORY_AND_DISK</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">2</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>MEMORY_AND_DISK_SER</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">1</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>MEMORY_AND_DISK_SER_2</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">2</p></td>
</tr>
<tr>
<td class="tableblock halign-left valign-top"><p class="tableblock"><code>OFF_HEAP</code></p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">true</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">false</p></td>
<td class="tableblock halign-left valign-top"><p class="tableblock">1</p></td>
</tr>
</tbody>
</table>
<div class="sidebarblock">
<div class="content">
<div class="title">如何选择分区级别</div>
<div class="paragraph">
<p>Spark 的存储级别的选择，核心问题是在 memory 内存使用率和 CPU 效率之间进行权衡。建议按下面的过程进行存储级别的选择:</p>
</div>
<div class="paragraph">
<p>如果您的 RDD 适合于默认存储级别（MEMORY_ONLY），leave them that way。这是 CPU 效率最高的选项，允许 RDD 上的操作尽可能快地运行.</p>
</div>
<div class="paragraph">
<p>如果不是，试着使用 MEMORY_ONLY_SER 和 selecting a fast serialization library 以使对象更加节省空间，但仍然能够快速访问。(Java和Scala)</p>
</div>
<div class="paragraph">
<p>不要溢出到磁盘，除非计算您的数据集的函数是昂贵的，或者它们过滤大量的数据。否则，重新计算分区可能与从磁盘读取分区一样快.</p>
</div>
<div class="paragraph">
<p>如果需要快速故障恢复，请使用复制的存储级别（例如，如果使用 Spark 来服务 来自网络应用程序的请求）。All 存储级别通过重新计算丢失的数据来提供完整的容错能力，但复制的数据可让您继续在 RDD 上运行任务，而无需等待重新计算一个丢失的分区.</p>
</div>
</div>
</div>
</div>
</div>
</div>
<div class="sect1">
<h2 id="_5_checkpoint">5. Checkpoint</h2>
<div class="sectionbody">
<div class="exampleblock">
<div class="title">目标</div>
<div class="content">
<div class="olist arabic">
<ol class="arabic">
<li>
<p>Checkpoint 的作用</p>
</li>
<li>
<p>Checkpoint 的使用</p>
</li>
</ol>
</div>
</div>
</div>
<div class="sect2">
<h3 id="_5_1_checkpoint_的作用">5.1. Checkpoint 的作用</h3>
<div class="paragraph">
<p>Checkpoint 的主要作用是斩断 RDD 的依赖链, 并且将数据存储在可靠的存储引擎中, 例如支持分布式存储和副本机制的 HDFS.</p>
</div>
<div class="dlist">
<dl>
<dt class="hdlist1">Checkpoint 的方式</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="ulist">
<ul>
<li>
<p><strong>可靠的</strong> 将数据存储在可靠的存储引擎中, 例如 HDFS</p>
</li>
<li>
<p><strong>本地的</strong> 将数据存储在本地</p>
</li>
</ul>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">什么是斩断依赖链</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="paragraph">
<p>斩断依赖链是一个非常重要的操作, 接下来以 HDFS 的 NameNode 的原理来举例说明</p>
</div>
<div class="paragraph">
<p>HDFS 的 NameNode 中主要职责就是维护两个文件, 一个叫做 <code>edits</code>, 另外一个叫做 <code>fsimage</code>. <code>edits</code> 中主要存放 <code>EditLog</code>, <code>FsImage</code> 保存了当前系统中所有目录和文件的信息. 这个 <code>FsImage</code> 其实就是一个 <code>Checkpoint</code>.</p>
</div>
<div class="paragraph">
<p>HDFS 的 NameNode 维护这两个文件的主要过程是, 首先, 会由 <code>fsimage</code> 文件记录当前系统某个时间点的完整数据, 自此之后的数据并不是时刻写入 <code>fsimage</code>, 而是将操作记录存储在 <code>edits</code> 文件中. 其次, 在一定的触发条件下, <code>edits</code> 会将自身合并进入 <code>fsimage</code>. 最后生成新的 <code>fsimage</code> 文件, <code>edits</code> 重置, 从新记录这次 <code>fsimage</code> 以后的操作日志.</p>
</div>
<div class="paragraph">
<p>如果不合并 <code>edits</code> 进入 <code>fsimage</code> 会怎样? 会导致 <code>edits</code> 中记录的日志过长, 容易出错.</p>
</div>
<div class="paragraph">
<p>所以当 Spark 的一个 Job 执行流程过长的时候, 也需要这样的一个斩断依赖链的过程, 使得接下来的计算轻装上阵.</p>
</div>
</div>
</div>
</dd>
<dt class="hdlist1">Checkpoint 和 Cache 的区别</dt>
<dd>
<div class="sidebarblock">
<div class="content">
<div class="paragraph">
<p>Cache 可以把 RDD 计算出来然后放在内存中, 但是 RDD 的依赖链(相当于 NameNode 中的 Edits 日志)是不能丢掉的, 因为这种缓存是不可靠的, 如果出现了一些错误(例如 Executor 宕机), 这个 RDD 的容错就只能通过回溯依赖链, 重放计算出来.</p>
</div>
<div class="paragraph">
<p>但是 Checkpoint 把结果保存在 HDFS 这类存储中, 就是可靠的了, 所以可以斩断依赖, 如果出错了, 则通过复制 HDFS 中的文件来实现容错.</p>
</div>
<div class="paragraph">
<p>所以他们的区别主要在以下两点</p>
</div>
<div class="ulist">
<ul>
<li>
<p>Checkpoint 可以保存数据到 HDFS 这类可靠的存储上, Persist 和 Cache 只能保存在本地的磁盘和内存中</p>
</li>
<li>
<p>Checkpoint 可以斩断 RDD 的依赖链, 而 Persist 和 Cache 不行</p>
</li>
<li>
<p>因为 CheckpointRDD 没有向上的依赖链, 所以程序结束后依然存在, 不会被删除. 而 Cache 和 Persist 会在程序结束后立刻被清除.</p>
</li>
</ul>
</div>
</div>
</div>
</dd>
</dl>
</div>
</div>
<div class="sect2">
<h3 id="_5_2_使用_checkpoint">5.2. 使用 Checkpoint</h3>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val conf = new SparkConf().setMaster("local[6]").setAppName("debug_string")
val sc = new SparkContext(conf)
sc.setCheckpointDir("checkpoint") <i class="conum" data-value="1"></i><b>(1)</b>

val interimRDD = sc.textFile("dataset/access_log_sample.txt")
  .map(item =&gt; (item.split(" ")(0), 1))
  .filter(item =&gt; StringUtils.isNotBlank(item._1))
  .reduceByKey((curr, agg) =&gt; curr + agg)

interimRDD.checkpoint() <i class="conum" data-value="2"></i><b>(2)</b>

interimRDD.collect().foreach(println(_))

sc.stop()</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td>在使用 Checkpoint 之前需要先设置 Checkpoint 的存储路径, 而且如果任务在集群中运行的话, 这个路径必须是 HDFS 上的路径</td>
</tr>
<tr>
<td><i class="conum" data-value="2"></i><b>2</b></td>
<td>开启 Checkpoint</td>
</tr>
</table>
</div>
<div class="admonitionblock note">
<table>
<tr>
<td class="icon">
<i class="fa icon-note" title="Note"></i>
</td>
<td class="content">
<div class="title">一个小细节</div>
<div class="listingblock">
<div class="content">
<pre class="highlightjs highlight"><code class="language-scala hljs" data-lang="scala">val interimRDD = sc.textFile("dataset/access_log_sample.txt")
  .map(item =&gt; (item.split(" ")(0), 1))
  .filter(item =&gt; StringUtils.isNotBlank(item._1))
  .reduceByKey((curr, agg) =&gt; curr + agg)
  .cache() <i class="conum" data-value="1"></i><b>(1)</b>

interimRDD.checkpoint()
interimRDD.collect().foreach(println(_))</code></pre>
</div>
</div>
<div class="colist arabic">
<table>
<tr>
<td><i class="conum" data-value="1"></i><b>1</b></td>
<td><code>checkpoint</code> 之前先 <code>cache</code> 一下, 准没错</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>应该在 <code>checkpoint</code> 之前先 <code>cache</code> 一下, 因为 <code>checkpoint</code> 会重新计算整个 RDD 的数据然后再存入 HDFS 等地方.</p>
</div>
<div class="paragraph">
<p>所以上述代码中如果 <code>checkpoint</code> 之前没有 <code>cache</code>, 则整个流程会被计算两次, 一次是 <code>checkpoint</code>, 另外一次是 <code>collect</code></p>
</div>
</td>
</tr>
</table>
</div>
<div class="paragraph">
<p>end</p>
</div>
</div>
</div>
</div>
        </div>
      </div>
    </body>
  </html>
